eBPF tracer: fix the Kafka protocol detection

ebpftracer/ebpf/l7/kafka.c

@@ -21,6 +21,14 @@ __s32 is_kafka_request(char *buf, int buf_size) {
     if (bpf_probe_read(&h, sizeof(h), (void *)((char *)buf)) < 0) {
         return 0;
     }
+    h.length = bpf_htonl(h.length);
+    h.api_key = bpf_htons(h.api_key);
+    h.api_version = bpf_htons(h.api_version);
+    h.correlation_id = bpf_htonl(h.correlation_id);
+
+    if (h.length+4 != buf_size) {
+        return 0;
+    }
     if (h.correlation_id > 0 && (h.api_key >= 0 && h.api_key <= 67)) {
         return h.correlation_id;
     }
@@ -33,7 +41,7 @@ __u32 parse_kafka_status(__s32 request_id, char *buf, int buf_size, __u8 partial
     if (bpf_probe_read(&h, sizeof(h), (void *)((char *)buf)) < 0) {
         return 0;
     }
-    if (h.correlation_id == request_id) {
+    if (bpf_htonl(h.correlation_id) == request_id) {
         return 200;
     }
     return 0;

ebpftracer/ebpf/l7/l7.c

@@ -117,10 +117,12 @@ int trace_enter_write(__u64 fd, char *buf, __u64 size) {
             if (prev_req && prev_req->protocol == PROTOCOL_KAFKA) {
                 req.ns = prev_req->ns;
             }
-        }
-        k.stream_id = is_cassandra_request(buf, size);
-        if  (k.stream_id != -1) {
-            req.protocol = PROTOCOL_CASSANDRA;
+        } else {
+            __s16 stream_id = is_cassandra_request(buf, size);
+            if  (stream_id != -1) {
+                k.stream_id = stream_id;
+                req.protocol = PROTOCOL_CASSANDRA;
+            }
         }
     }
     if (req.protocol == PROTOCOL_UNKNOWN) {

ebpftracer/ebpf/l7/postgres.c

@@ -15,7 +15,7 @@ int is_postgres_query(char *buf, int buf_size) {
     if (bpf_probe_read(&f_length, sizeof(f_length), (void *)((char *)buf+1)) < 0) {
         return 0;
     }
-    f_length = bpf_ntohl(f_length);
+    f_length = bpf_htonl(f_length);
     if (f_cmd == 'Q' && f_length+1 == buf_size) {
         return 1;
     }
@@ -39,7 +39,7 @@ __u32 parse_postgres_status(char *buf, int buf_size) {
     if (bpf_probe_read(&length, sizeof(length), (void *)((char *)buf+1)) < 0) {
         return 0;
     }
-    length = bpf_ntohl(length);
+    length = bpf_htonl(length);
 
     if (length+1 > buf_size) {
         return 0;