eBPF tracer: cassandra protocol support

Author: def

containers/metrics.go

@@ -81,6 +81,7 @@ var (
 		ebpftracer.L7ProtocolMysql:     {Name: "container_mysql_queries_total", Help: "Total number of outbound Mysql queries"},
 		ebpftracer.L7ProtocolMongo:     {Name: "container_mongo_queries_total", Help: "Total number of outbound Mongo queries"},
 		ebpftracer.L7ProtocolKafka:     {Name: "container_kafka_requests_total", Help: "Total number of outbound Kafka requests"},
+		ebpftracer.L7ProtocolCassandra: {Name: "container_cassandra_queries_total", Help: "Total number of outbound Cassandra requests"},
 	}
 	L7Latency = map[ebpftracer.L7Protocol]prometheus.HistogramOpts{
 		ebpftracer.L7ProtocolHTTP:      {Name: "container_http_requests_duration_seconds_total", Help: "Histogram of the response time for each outbound HTTP request"},
@@ -90,6 +91,7 @@ var (
 		ebpftracer.L7ProtocolMysql:     {Name: "container_mysql_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Mysql query"},
 		ebpftracer.L7ProtocolMongo:     {Name: "container_mongo_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Mongo query"},
 		ebpftracer.L7ProtocolKafka:     {Name: "container_kafka_requests_duration_seconds_total", Help: "Histogram of the execution time for each outbound Kafka request"},
+		ebpftracer.L7ProtocolCassandra: {Name: "container_cassandra_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Cassandra request"},
 	}
 )
 

ebpftracer/ebpf.go

@@ -0,0 +1,44 @@
+// https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec
+
+#define CASSANDRA_REQUEST_FRAME  0x04
+#define CASSANDRA_RESPONSE_FRAME 0x84
+
+#define CASSANDRA_OPCODE_ERROR  0x00
+#define CASSANDRA_OPCODE_RESULT 0x08
+
+struct cassandra_header {
+    __u8 version;
+    __u8 flags;
+    __s16 stream_id;
+    __u8 opcode;
+};
+
+static __always_inline
+__s16 is_cassandra_request(char *buf, int buf_size) {
+    if (buf_size < 1) {
+        return -1;
+    }
+    struct cassandra_header h = {};
+    if (bpf_probe_read(&h, sizeof(h), (void *)buf) < 0) {
+        return -1;
+    }
+    if (h.version == CASSANDRA_REQUEST_FRAME && h.stream_id >= 0) {
+        return h.stream_id;
+    }
+    return -1;
+}
+
+static __always_inline
+__u32 cassandra_status(struct cassandra_header h) {
+    if (h.version != CASSANDRA_RESPONSE_FRAME || h.stream_id == -1) {
+        return 0;
+    }
+    if (h.opcode == CASSANDRA_OPCODE_RESULT) {
+        return 200;
+    }
+    if (h.opcode == CASSANDRA_OPCODE_ERROR) {
+        return 500;
+    }
+    return 0;
+}
+

ebpftracer/ebpf/l7/cassandra.c

@@ -5,6 +5,7 @@
 #include "mysql.c"
 #include "mongo.c"
 #include "kafka.c"
+#include "cassandra.c"
 
 
 #define PROTOCOL_UNKNOWN    0
@@ -15,6 +16,7 @@
 #define PROTOCOL_MYSQL      5
 #define PROTOCOL_MONGO      6
 #define PROTOCOL_KAFKA      7
+#define PROTOCOL_CASSANDRA  8
 
 struct l7_event {
     __u64 fd;
@@ -46,6 +48,7 @@ struct {
 struct socket_key {
     __u64 fd;
     __u32 pid;
+    __s16 stream_id;
 };
 
 struct l7_request {
@@ -92,6 +95,7 @@ int trace_enter_write(__u64 fd, char *buf, __u64 size) {
     struct socket_key k = {};
     k.pid = id >> 32;
     k.fd = fd;
+    k.stream_id = -1;
     if (is_http_request(buf)) {
         req.protocol = PROTOCOL_HTTP;
     } else if (is_postgres_query(buf, size)) {
@@ -114,6 +118,10 @@ int trace_enter_write(__u64 fd, char *buf, __u64 size) {
                 req.ns = prev_req->ns;
             }
         }
+        k.stream_id = is_cassandra_request(buf, size);
+        if  (k.stream_id != -1) {
+            req.protocol = PROTOCOL_CASSANDRA;
+        }
     }
     if (req.protocol == PROTOCOL_UNKNOWN) {
         return 0;
@@ -147,16 +155,26 @@ int trace_exit_read(struct trace_event_raw_sys_exit_rw__stub* ctx) {
     struct socket_key k = {};
     k.pid = id >> 32;
     k.fd = args->fd;
+    k.stream_id = -1;
     buf = args->buf;
 
     bpf_map_delete_elem(&active_reads, &id);
     if (ctx->ret <= 0) {
         return 0;
     }
+    struct cassandra_header cassandra_response = {};
+    cassandra_response.stream_id = -1;
 
     struct l7_request *req = bpf_map_lookup_elem(&active_l7_requests, &k);
     if (!req) {
-        return 0;
+        if (bpf_probe_read(&cassandra_response, sizeof(cassandra_response), (void *)(buf)) < 0) {
+            return 0;
+        }
+        k.stream_id = cassandra_response.stream_id;
+        req = bpf_map_lookup_elem(&active_l7_requests, &k);
+        if (!req) {
+            return 0;
+        }
     }
     __s32 request_id = req->request_id;
     struct l7_event e = {};
@@ -189,6 +207,8 @@ int trace_exit_read(struct trace_event_raw_sys_exit_rw__stub* ctx) {
         }
     } else if (e.protocol == PROTOCOL_KAFKA) {
         e.status = parse_kafka_status(request_id, buf, ctx->ret, partial);
+    } else if (e.protocol == PROTOCOL_CASSANDRA) {
+        e.status = cassandra_status(cassandra_response);
     }
     if (e.status == 0) {
         return 0;

ebpftracer/ebpf/l7/l7.c

@@ -49,6 +49,7 @@ const (
 	L7ProtocolMysql     L7Protocol = 5
 	L7ProtocolMongo     L7Protocol = 6
 	L7ProtocolKafka     L7Protocol = 7
+	L7ProtocolCassandra L7Protocol = 8
 )
 
 func (p L7Protocol) String() string {
@@ -67,6 +68,8 @@ func (p L7Protocol) String() string {
 		return "Mongo"
 	case L7ProtocolKafka:
 		return "Kafka"
+	case L7ProtocolCassandra:
+		return "Cassandra"
 	}
 	return "UNKNOWN:" + strconv.Itoa(int(p))
 }