ebpf/ebpftracer/ebpf/l7/l7.c

#define PROTOCOL_UNKNOWN    0
#define PROTOCOL_HTTP	    1
#define PROTOCOL_POSTGRES	2
#define PROTOCOL_REDIS	    3
#define PROTOCOL_MEMCACHED  4
#define PROTOCOL_MYSQL      5
#define PROTOCOL_MONGO      6
#define PROTOCOL_KAFKA      7
#define PROTOCOL_CASSANDRA  8
#define PROTOCOL_RABBITMQ   9
#define PROTOCOL_NATS      10
#define PROTOCOL_HTTP2	   11
#define PROTOCOL_DUBBO2    12

#define PROTOCOL_TRACE	   200

#define STATUS_UNKNOWN  0
#define STATUS_OK       200
#define STATUS_FAILED   500

#define METHOD_UNKNOWN              0
#define METHOD_PRODUCE              1
#define METHOD_CONSUME              2
#define METHOD_STATEMENT_PREPARE    3
#define METHOD_STATEMENT_CLOSE      4
#define METHOD_HTTP2_CLIENT_FRAMES  5
#define METHOD_HTTP2_SERVER_FRAMES  6

#define MAX_PAYLOAD_SIZE 1024 // must be power of 2
#define TRUNCATE_PAYLOAD_SIZE(size) ({                                  \
    size = MIN(size, MAX_PAYLOAD_SIZE-1);                               \
    asm volatile ("%0 &= %1" : "+r"(size) : "i"(MAX_PAYLOAD_SIZE-1));   \
})
#define COPY_PAYLOAD(dst, size, src) ({     \
    TRUNCATE_PAYLOAD_SIZE(size);            \
    if (bpf_probe_read(dst, size, src)) {   \
        return 0;                           \
    }                                       \
})
#define NS_PER_SEC		1000000000ULL

#define IOVEC_BUF_SIZE MAX_PAYLOAD_SIZE * 2  // must be double of MAX_PAYLOAD_SIZE
#define MAX_IOVEC_SIZE 32

#include "http.c"
#include "postgres.c"
#include "redis.c"
#include "memcached.c"
#include "mysql.c"
#include "mongo.c"
#include "kafka.c"
#include "cassandra.c"
#include "rabbitmq.c"
#include "nats.c"
#include "http2.c"
#include "dubbo2.c"
#include "apm_trace.c"


struct l7_event {
    __u64 fd;
    __u64 connection_timestamp;
    __u32 pid;
    __u32 status;
    __u64 duration;
    __u8 protocol;
    __u8 method;
    __u16 padding;
    __u32 statement_id;
    __u64 payload_size;
    __u64 trace_id;
    __u32 trace_start;
    __u32 trace_end;
//    __u32 test_id;
    char payload[MAX_PAYLOAD_SIZE];
};

struct test_t {
    __u32 test_id;
};

struct {
     __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
     __type(key, int);
     __type(value, struct l7_event);
     __uint(max_entries, 1);
} l7_event_heap SEC(".maps");

struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
    __type(key, int);
    __type(value, struct test_t);
    __uint(max_entries, 1);
} test_heap SEC(".maps");

struct {
    __uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
    __uint(key_size, sizeof(int));
    __uint(value_size, sizeof(int));
} l7_events SEC(".maps");

struct read_args {
    __u64 fd;
    char* buf;
    __u64* ret;
    __u64 iovlen;
};

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(key_size, sizeof(__u64));
    __uint(value_size, sizeof(struct read_args));
    __uint(max_entries, 10240);
} active_reads SEC(".maps");

struct l7_request_key {
    __u64 fd;
    __u32 pid;
    __u16 is_tls;
    __s16 stream_id;
};

struct l7_request {
    __u64 ns;
    __u8 protocol;
    __u8 partial;
    __u8 request_type;
    __s32 request_id;
    __u64 payload_size;
    char payload[MAX_PAYLOAD_SIZE];
};

struct {
     __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
     __type(key, int);
     __type(value, struct l7_request);
     __uint(max_entries, 1);
} l7_request_heap SEC(".maps");

struct {
    __uint(type, BPF_MAP_TYPE_LRU_HASH);
    __uint(key_size, sizeof(struct l7_request_key));
    __uint(value_size, sizeof(struct l7_request));
    __uint(max_entries, 32768);
} active_l7_requests SEC(".maps");

struct {
     __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
     __type(key, int);
     __type(value, char[IOVEC_BUF_SIZE]);
     __uint(max_entries, 1);
} iovec_buf_heap SEC(".maps");

struct trace_event_raw_sys_enter_rw__stub {
    __u64 unused;
    long int id;
    __u64 fd;
    char* buf;
    __u64 size;
};

struct trace_event_raw_sys_exit_rw__stub {
    __u64 unused;
    long int id;
    long int ret;
};

struct l7_iovec {
    char* buf;
    __u64 size;
};

struct l7_user_msghdr {
	void *msg_name;
	int msg_namelen;
	struct l7_iovec *msg_iov;
	__u64 msg_iovlen;
};

static inline __attribute__((__always_inline__))
void send_event(void *ctx, struct l7_event *e, __u32 pid, __u64 fd) {
    struct sk_info sk = {};
    sk.pid = pid;
    sk.fd = fd;
    __u64 *timestamp = bpf_map_lookup_elem(&connection_timestamps, &sk);
    if (timestamp) {
        if (*timestamp == 0) {
            return;
        }
        e->connection_timestamp = *timestamp;
    } else {
        e->connection_timestamp = 0;
    }
    e->fd = fd;
    e->pid = pid;
    bpf_perf_event_output(ctx, &l7_events, BPF_F_CURRENT_CPU, e, sizeof(*e));
}

static inline __attribute__((__always_inline__))
__u64 read_iovec(char *l7_iovec, __u64 iovlen, __u64 ret, char *buf) {
    struct l7_iovec iov = {};
    __u64 max = (ret) ? MIN(ret, MAX_PAYLOAD_SIZE) : MAX_PAYLOAD_SIZE;
    __u64 offset = 0;
    __u64 size = 0;
    #pragma unroll
    for (int i = 0; i < MAX_IOVEC_SIZE; i++) {
        if (i >= iovlen) {
            break;
        }
        if (bpf_probe_read(&iov, sizeof(iov), (void *)(l7_iovec+i*sizeof(iov)))) {
            return 0;
        }
        if (iov.size <= 0) {
            continue;
        }
        size = MIN(iov.size, max-offset);
        TRUNCATE_PAYLOAD_SIZE(size);
        TRUNCATE_PAYLOAD_SIZE(offset);
        if (bpf_probe_read(buf + offset, size, (void *)iov.buf)) {
            return 0;
        }
        offset += size;
        if (offset >= max) {
            break;
        }
    }
    return offset;
}

static inline __attribute__((__always_inline__))
int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size, __u64 iovlen) {
    __u64 id = bpf_get_current_pid_tgid();
    __u32 zero = 0;
    __u32 pid, tid;
    __u32 http_status ;

    pid = id >> 32;
    tid =  (__u32)id;
    if (load_filter_pid() != 0 && pid != load_filter_pid()) {
        return 0;
    }
    char* payload = buf;
    if (iovlen) {
        payload = bpf_map_lookup_elem(&iovec_buf_heap, &zero);
        if (!payload) {
            return 0;
        }
        size = read_iovec(buf, iovlen, 0, payload);
    }
    if (!size) {
        return 0;
    }

    struct l7_request *req = bpf_map_lookup_elem(&l7_request_heap, &zero);
    if (!req) {
        return 0;
    }
    req->protocol = PROTOCOL_UNKNOWN;
    req->partial = 0;
    req->request_id = 0;
    req->ns = 0;
    req->payload_size = size;
    struct l7_request_key k = {};
    k.pid = id >> 32;
    k.fd = fd;
    k.is_tls = is_tls;
    k.stream_id = -1;

//    cw_bpf_debug("enter-payload:%s|type:%s|FD:%d\n",payload,"type",k.fd);

    if (is_http_response(payload, &http_status))
    {
//        __u64 goid = get_rw_goid(120 * NS_PER_SEC, 1);
//        cw_bpf_debug("[Response][HTTP]:thread_id:%d|goid:%d|FD:%d\n", tid, goid, k.fd);
//        struct trace_key_t trace_key = get_trace_key(pid, tid);
//        struct fd_trace_key_t fd_trace_key = get_fd_trace_key(pid, fd);
        __u64 trace_id = get_fd_trace_id(pid, fd);
//        cw_bpf_debug("trace_id:%llu", trace_id);
        cw_bpf_debug("[trace end][Response][HTTP] pid:%d,fd:%d,traceid:%llu", pid, fd, trace_id);
        // 清除trace信息
        clear_trace(pid, tid, fd);
        // 发送事件到用户空间 start
        struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
        if (!e) {
            return 0;
        }
        struct l7_request *req = bpf_map_lookup_elem(&active_l7_requests, &k);
        if (!req)
        {
//            cw_bpf_debug("[Response][HTTP]:no req-----------");
//            cw_bpf_debug("[Response][HTTP]:pid:%d|tid:%d",k.pid,k.fd);
//            cw_bpf_debug("[Response][HTTP]:is_tls:%d|tid:%d",k.is_tls,k.stream_id);
            return 0;
        }

        e->duration = bpf_ktime_get_ns() - req->ns;
//        cw_bpf_debug("[Response][HTTP]:duration->ns:%d\n",e->duration);
        e->protocol = PROTOCOL_TRACE;
        e->status = http_status;
        e->pid = k.pid;
        e->fd = k.fd;
        // e->connection_timestamp = get_connection_timestamp(k.pid, k.fd);
        e->trace_start = 0;
        e->trace_end = 1;
        e->trace_id = trace_id;
        e->payload_size = size;
        COPY_PAYLOAD(e->payload, size, payload);
        bpf_map_delete_elem(&active_l7_requests, &k);
        bpf_perf_event_output(ctx, &l7_events, BPF_F_CURRENT_CPU, e, sizeof(*e));
        // 发送事件到用户空间 end
        __u64 k_version = load_filter_pid();
        cw_bpf_debug("filter_pid:%d", k_version);

        struct test_t *ttt = bpf_map_lookup_elem(&test_heap, &zero);
        if (!ttt) {
            return 0;
        }

//        struct member_fields_offset *off = bpf_map_lookup_elem(&__members_offset, &zero);
//        if (!off) {
//            return 0;
//        }

//        cw_bpf_debug("off->task__files_offset:%x", off->task__files_offset);
        cw_bpf_debug("e->test_id:%d", ttt->test_id);
        cw_bpf_debug("HTTP_END");
        return 0;
    }

    if (is_http_request(payload)) {
        cw_bpf_debug("[Enter][HTTP222]:TGID:%d|FD:%d\n",k.pid,k.fd);
        req->protocol = PROTOCOL_HTTP;
    } else if (is_postgres_query(payload, size, &req->request_type)) {
        if (req->request_type == POSTGRES_FRAME_CLOSE) {
            struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
            if (!e) {
                return 0;
            }
            e->protocol = PROTOCOL_POSTGRES;
            e->method = METHOD_STATEMENT_CLOSE;
            e->payload_size = size;
            COPY_PAYLOAD(e->payload, size, payload);
            send_event(ctx, e, k.pid, k.fd);
            return 0;
        }
        req->protocol = PROTOCOL_POSTGRES;
    } else if (is_redis_query(payload, size)) {
        cw_bpf_debug("[Enter][Redis]:TGID:%d|type:%s|FD:%d\n",k.pid,"type",k.fd);
        req->protocol = PROTOCOL_REDIS;
    } else if (is_memcached_query(payload, size)) {
        req->protocol = PROTOCOL_MEMCACHED;
    } else if (is_mysql_query(payload, size, &req->request_type)) {
        cw_bpf_debug("[Enter][Mysql]:thread_id:%d\n",tid);
        if (req->request_type == MYSQL_COM_STMT_CLOSE) {
            struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
            if (!e) {
                return 0;
            }
            e->protocol = PROTOCOL_MYSQL;
            e->method = METHOD_STATEMENT_CLOSE;
            e->payload_size = size;
            COPY_PAYLOAD(e->payload, size, payload);
            send_event(ctx, e, k.pid, k.fd);
            return 0;
        }
        req->protocol = PROTOCOL_MYSQL;
    } else if (is_mongo_query(payload, size)) {
        req->protocol = PROTOCOL_MONGO;
    } else if (is_rabbitmq_produce(payload, size)) {
        struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
        if (!e) {
            return 0;
        }
        e->protocol = PROTOCOL_RABBITMQ;
        e->method = METHOD_PRODUCE;
        send_event(ctx, e, k.pid, k.fd);
        return 0;
    } else if (nats_method(payload, size) == METHOD_PRODUCE) {
        struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
        if (!e) {
            return 0;
        }
        e->protocol = PROTOCOL_NATS;
        e->method = METHOD_PRODUCE;
        send_event(ctx, e, k.pid, k.fd);
        return 0;
    } else if (is_cassandra_request(payload, size, &k.stream_id)) {
        req->protocol = PROTOCOL_CASSANDRA;
    } else if (is_kafka_request(payload, size, &req->request_id)) {
        req->protocol = PROTOCOL_KAFKA;
        struct l7_request *prev_req = bpf_map_lookup_elem(&active_l7_requests, &k);
        if (prev_req && prev_req->protocol == PROTOCOL_KAFKA) {
            req->ns = prev_req->ns;
        }
    } else if (looks_like_http2_frame(payload, size, METHOD_HTTP2_CLIENT_FRAMES)) {
        struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
        if (!e) {
            return 0;
        }
        e->protocol = PROTOCOL_HTTP2;
        e->method = METHOD_HTTP2_CLIENT_FRAMES;
        e->duration = bpf_ktime_get_ns();
        e->payload_size = size;
        COPY_PAYLOAD(e->payload, size, payload);
        send_event(ctx, e, k.pid, k.fd);
        return 0;
    } else if (is_dubbo2_request(payload, size)) {
        req->protocol = PROTOCOL_DUBBO2;
    }

    if (req->protocol == PROTOCOL_UNKNOWN) {
        return 0;
    }
    if (req->ns == 0) {
        req->ns = bpf_ktime_get_ns();
    }
    COPY_PAYLOAD(req->payload, size, payload);
    bpf_map_update_elem(&active_l7_requests, &k, req, BPF_NOEXIST);
    return 0;
}

static inline __attribute__((__always_inline__))
int trace_enter_read(__u64 id, __u64 fd, char *buf, __u64 *ret, __u64 iovlen) {
    struct read_args args = {};
    args.fd = fd;
    args.buf = buf;
    args.ret = ret;
    args.iovlen = iovlen;
    __u32 pid = id >> 32;
    bpf_map_update_elem(&active_reads, &id, &args, BPF_ANY);
    return 0;
}

static inline __attribute__((__always_inline__))
int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret) {
    __u32 tid =  (__u32)id;

    if (load_filter_pid() != 0 && pid != load_filter_pid()) {
        return 0;
    }

    struct read_args *args = bpf_map_lookup_elem(&active_reads, &id);
    if (!args) {
        return 0;
    }

    struct l7_request_key k = {};
    k.pid = pid;
    k.fd = args->fd;
    k.is_tls = is_tls;
    k.stream_id = -1;

    bpf_map_delete_elem(&active_reads, &id);

    if (ret <= 0) {
        return 0;
    }
    if (args->ret) {
        if (bpf_probe_read(&ret, sizeof(ret), (void*)args->ret)) {
            return 0;
        };
        if (ret <= 0) {
            return 0;
        }
    }

    int zero = 0;
    char* payload = args->buf;
    if (args->iovlen) {
        payload = bpf_map_lookup_elem(&iovec_buf_heap, &zero);
        if (!payload) {
            return 0;
        }
        ret = read_iovec(args->buf, args->iovlen, ret, payload);
        if (!ret) {
            return 0;
        }
    }

    struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
    if (!e) {
        return 0;
    }
    e->fd = k.fd;
    e->pid = k.pid;
    e->protocol = PROTOCOL_UNKNOWN;
    e->status = STATUS_UNKNOWN;
    e->method = METHOD_UNKNOWN;
    e->statement_id = 0;
    e->payload_size = 0;

//    __u32 k0 = 0;
//    struct member_fields_offset *offset = members_offset__lookup(&k0);
//    if (!offset)
//        return -1;
//    void *sk = get_socket_from_fd(args->fd, offset);
//    struct conn_info_t *conn_info, __conn_info = { 0 };
//    conn_info = &__conn_info;
////
//    init_conn_info(id >> 32, args->fd, &__conn_info, sk, offset);
////
//    infer_dns_message(payload, (int)PT_REGS_RC((struct pt_regs *)ctx),
//                      conn_info);

    // 被调用方http入口
    if (is_http_request(payload)) {
        struct l7_request *req = bpf_map_lookup_elem(&l7_request_heap, &zero);
        if (!req)
        {
            return 0;
        } else
        {
            req->protocol = PROTOCOL_HTTP;
            req->payload_size = ret;
            req->ns = bpf_ktime_get_ns();
            COPY_PAYLOAD(req->payload, ret, payload);
            // cw_bpf_debug("[Receive][HTTP]:pid:%d|tid:%d",k.pid,k.fd);
            // cw_bpf_debug("[Receive][HTTP]:is_tls:%d|tid:%d",k.is_tls,k.stream_id);
            bpf_map_update_elem(&active_l7_requests, &k, req, BPF_NOEXIST);
        }

        cw_bpf_debug("[trace start][Receive][HTTP]:thread_id:%d\n",tid);
//        cw_bpf_debug("[Receive][HTTP] payload1:%s|type:%s\n",payload,"type");
//        __u64 goid = 0;
//        goid = get_rw_goid(120 * NS_PER_SEC, 1);
//        cw_bpf_debug("[Receive][HTTP] goid:%llu\n",goid);
//        __u64 goid = get_rw_goid(120 * NS_PER_SEC, 1);
//        cw_bpf_debug("[Receive][HTTP]:thread_id:%d|goid:%d|FD:%d\n", tid, goid, k.fd);

        struct apm_trace_key_t trace_key = {0};
        struct apm_trace_info_t trace_info = {0};
        trace_key.tgid = pid;
        trace_key.pid = tid;
        __u64 uid_base = bpf_ktime_get_ns();
        trace_info.trace_id = bpf_get_current_pid_tgid() + uid_base;
        e->trace_start = 1;
        e->trace_end = 0;
        e->protocol = PROTOCOL_TRACE;
        e->trace_id = trace_info.trace_id;
        e->payload_size = ret;
        COPY_PAYLOAD(e->payload, ret, payload);

        // http trace
        struct fd_trace_key_t fd_trace_key = {};
        fd_trace_key.tgid = pid;
        fd_trace_key.fd = k.fd;
        bpf_map_update_elem(&fd_trace_info_heap, &fd_trace_key, &trace_info, BPF_NOEXIST);
        cw_bpf_debug("[Receive][HTTP] pid:%d,fd:%d,traceid:%llu", fd_trace_key.tgid, fd_trace_key.fd,
                   trace_info.trace_id);

        // 入口方法缓存  bpf_map_update_elem(map, key, value, options)
        bpf_map_update_elem(&trace_info_heap, &trace_key, &trace_info, BPF_NOEXIST);
        bpf_perf_event_output(ctx, &l7_events, BPF_F_CURRENT_CPU, e, sizeof(*e));
        cw_bpf_debug("[Receive][HTTP] to user space");
        return 0;
    }

    if (is_rabbitmq_consume(payload, ret)) {
        e->protocol = PROTOCOL_RABBITMQ;
        e->method = METHOD_CONSUME;
        send_event(ctx, e, k.pid, k.fd);
        return 0;
    }
    if (nats_method(payload, ret) == METHOD_CONSUME) {
        e->protocol = PROTOCOL_NATS;
        e->method = METHOD_CONSUME;
        send_event(ctx, e, k.pid, k.fd);
        return 0;
    }

    struct l7_request *req = bpf_map_lookup_elem(&active_l7_requests, &k);
    int response = 0;
    if (!req) {
        if (is_cassandra_response(payload, ret, &k.stream_id, &e->status)) {
            req = bpf_map_lookup_elem(&active_l7_requests, &k);
            if (!req) {
                return 0;
            }
            response = 1;
        } else if (looks_like_http2_frame(payload, ret, METHOD_HTTP2_SERVER_FRAMES)) {
            e->protocol = PROTOCOL_HTTP2;
            e->method = METHOD_HTTP2_SERVER_FRAMES;
            e->duration = bpf_ktime_get_ns();
            e->payload_size = ret;
            COPY_PAYLOAD(e->payload, ret, payload);
            send_event(ctx, e, k.pid, k.fd);
            return 0;
        } else {
            return 0;
        }
    }

    e->protocol = req->protocol;
    e->payload_size = req->payload_size;
    COPY_PAYLOAD(e->payload, req->payload_size, req->payload);

    bpf_map_delete_elem(&active_l7_requests, &k);
    if (e->protocol == PROTOCOL_HTTP) {
        __u64 trace_id = get_trace_id(pid, tid);
        e->trace_id = trace_id;
        cw_bpf_debug("[Response][HTTP222]:thread_id:%d|type:%s|FD:%d\n",k.pid,"",k.fd);
        cw_bpf_debug("[Response][HTTP222] trace_id:%llu", trace_id);
        // 请求报文
        cw_bpf_debug("[Response][HTTP222] req-payload:%s",e->payload);
        // 响应报文
        cw_bpf_debug("[Response][HTTP222] resp-payload:%s",payload);

        response = is_http_response(payload, &e->status);
    } else if (e->protocol == PROTOCOL_POSTGRES) {
        response = is_postgres_response(payload, ret, &e->status);
        if (req->request_type == POSTGRES_FRAME_PARSE) {
            e->method = METHOD_STATEMENT_PREPARE;
        }
    } else if (e->protocol == PROTOCOL_REDIS) {
        cw_bpf_debug("[Response][Redis]:TGID:%d|type:%s|FD:%d\n",k.pid,"",k.fd);
        __u64 trace_id = get_trace_id(pid, tid);
        cw_bpf_debug("[Redis] trace_id:%llu", trace_id);
        e->trace_id = trace_id;
        response = is_redis_response(payload, ret, &e->status);
    } else if (e->protocol == PROTOCOL_MEMCACHED) {
        response = is_memcached_response(payload, ret, &e->status);
    } else if (e->protocol == PROTOCOL_MYSQL) {
        cw_bpf_debug("[Response][Mysql]:thread_id:%d\n",tid);
        __u64 trace_id = get_trace_id(pid, tid);
//        cw_bpf_debug("[Mysql] trace_id:%llu", trace_id);
        e->trace_id = trace_id;
        response = is_mysql_response(payload, ret, req->request_type, &e->statement_id, &e->status);
        if (req->request_type == MYSQL_COM_STMT_PREPARE) {
            e->method = METHOD_STATEMENT_PREPARE;
        }
    } else if (e->protocol == PROTOCOL_MONGO) {
        response = is_mongo_response(payload, ret, req->partial);
        if (response == 2) { // partial
            struct l7_request *r = bpf_map_lookup_elem(&l7_request_heap, &zero);
            if (!r) {
                return 0;
            }
            r->partial = 1;
            r->protocol = e->protocol;
            r->ns = req->ns;
            r->payload_size = req->payload_size;
            COPY_PAYLOAD(r->payload, req->payload_size, req->payload);
            bpf_map_update_elem(&active_l7_requests, &k, r, BPF_ANY);
            return 0;
        }
    } else if (e->protocol == PROTOCOL_KAFKA) {
        response = is_kafka_response(payload, req->request_id);
    } else if (e->protocol == PROTOCOL_DUBBO2) {
        response = is_dubbo2_response(payload, &e->status);
    }

    if (!response) {
        return 0;
    }
    e->duration = bpf_ktime_get_ns() - req->ns;
    send_event(ctx, e, k.pid, k.fd);
    return 0;
}

SEC("tracepoint/syscalls/sys_enter_write")
int sys_enter_write(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    return trace_enter_write(ctx, ctx->fd, 0, ctx->buf, ctx->size, 0);
}

SEC("tracepoint/syscalls/sys_enter_writev")
int sys_enter_writev(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    return trace_enter_write(ctx, ctx->fd, 0, ctx->buf, 0, ctx->size);
}

SEC("tracepoint/syscalls/sys_enter_sendmsg")
int sys_enter_sendmsg(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    struct l7_user_msghdr msghdr = {};
    if (bpf_probe_read(&msghdr, sizeof(msghdr), (void *)ctx->buf)) {
        return 0;
    }
    return trace_enter_write(ctx, ctx->fd, 0, (char*)msghdr.msg_iov, 0, msghdr.msg_iovlen);
}

SEC("tracepoint/syscalls/sys_enter_sendto")
int sys_enter_sendto(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    return trace_enter_write(ctx, ctx->fd, 0, ctx->buf, ctx->size, 0);
}

SEC("tracepoint/syscalls/sys_enter_read")
int sys_enter_read(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    __u64 id = bpf_get_current_pid_tgid();
    return trace_enter_read(id, ctx->fd, ctx->buf, 0, 0);
}

SEC("tracepoint/syscalls/sys_enter_readv")
int sys_enter_readv(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    __u64 id = bpf_get_current_pid_tgid();
    return trace_enter_read(id, ctx->fd, ctx->buf, 0, ctx->size);
}

SEC("tracepoint/syscalls/sys_enter_recvmsg")
int sys_enter_recvmsg(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    __u64 id = bpf_get_current_pid_tgid();
    struct l7_user_msghdr msghdr = {};
    if (bpf_probe_read(&msghdr, sizeof(msghdr), (void *)ctx->buf)) {
        return 0;
    }
    return trace_enter_read(id, ctx->fd, (char*)msghdr.msg_iov, 0, msghdr.msg_iovlen);
}

SEC("tracepoint/syscalls/sys_enter_recvfrom")
int sys_enter_recvfrom(struct trace_event_raw_sys_enter_rw__stub* ctx) {
    __u64 id = bpf_get_current_pid_tgid();
    return trace_enter_read(id, ctx->fd, ctx->buf, 0, 0);
}

SEC("tracepoint/syscalls/sys_exit_read")
int sys_exit_read(struct trace_event_raw_sys_exit_rw__stub* ctx) {
    __u64 pid_tgid = bpf_get_current_pid_tgid();
    __u32 pid = pid_tgid >> 32;
    return trace_exit_read(ctx, pid_tgid, pid, 0, ctx->ret);
}

SEC("tracepoint/syscalls/sys_exit_readv")
int sys_exit_readv(struct trace_event_raw_sys_exit_rw__stub* ctx) {
    __u64 pid_tgid = bpf_get_current_pid_tgid();
    __u32 pid = pid_tgid >> 32;
    return trace_exit_read(ctx, pid_tgid, pid, 0, ctx->ret);
}

SEC("tracepoint/syscalls/sys_exit_recvmsg")
int sys_exit_recvmsg(struct trace_event_raw_sys_exit_rw__stub* ctx) {
    __u64 pid_tgid = bpf_get_current_pid_tgid();
    __u32 pid = pid_tgid >> 32;
    return trace_exit_read(ctx, pid_tgid, pid, 0, ctx->ret);
}

SEC("tracepoint/syscalls/sys_exit_recvfrom")
int sys_exit_recvfrom(struct trace_event_raw_sys_exit_rw__stub* ctx) {
    __u64 pid_tgid = bpf_get_current_pid_tgid();
    __u32 pid = pid_tgid >> 32;
    return trace_exit_read(ctx, pid_tgid, pid, 0, ctx->ret);
}
//
//SEC("tracepoint/syscalls/sys_exit_recvfrom")
//int sys_exit_recvfrom222(struct trace_event_raw_sys_exit_rw__stub* ctx) {
//    __u64 pid_tgid = bpf_get_current_pid_tgid();
//    __u32 pid = pid_tgid >> 32;
//    if (pid == filterPid) {
//        cw_bpf_debug("sys_exit_recvfrom222");
//    }
//    return 0;
//}