capturing Mongo queries at the eBPF level

Author: def

containers/metrics.go

@@ -79,13 +79,15 @@ var (
 		ebpftracer.L7ProtocolRedis:     {Name: "container_redis_queries_total", Help: "Total number of outbound Redis queries"},
 		ebpftracer.L7ProtocolMemcached: {Name: "container_memcached_queries_total", Help: "Total number of outbound Memcached queries"},
 		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"},
 	}
 	L7Latency = map[ebpftracer.L7Protocol]prometheus.HistogramOpts{
 		ebpftracer.L7ProtocolHTTP:      {Name: "container_http_request_duration_seconds_total", Help: "Histogram of the response time for each outbound HTTP request"},
 		ebpftracer.L7ProtocolPostgres:  {Name: "container_postgres_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Postgres query"},
 		ebpftracer.L7ProtocolRedis:     {Name: "container_redis_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Redis query"},
 		ebpftracer.L7ProtocolMemcached: {Name: "container_memcached_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Memcached query"},
 		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/ebpf.go

@@ -3,13 +3,15 @@
 #include "redis.c"
 #include "memcached.c"
 #include "mysql.c"
+#include "mongo.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
 
 struct l7_event {
 	__u64 fd;
@@ -46,6 +48,7 @@ struct socket_key {
 struct l7_request {
     __u64 ns;
     __u8 protocol;
+    __u8 partial;
 };
 
 struct {
@@ -80,6 +83,7 @@ static inline __attribute__((__always_inline__))
 int trace_enter_write(__u64 fd, char *buf, __u64 size) {
     __u32 pid = bpf_get_current_pid_tgid() >> 32;
     struct l7_request req = {};
+    req.partial = 0;
     if (is_http_request(buf)) {
         req.protocol = PROTOCOL_HTTP;
     } else if (is_postgres_query(buf, size)) {
@@ -90,6 +94,8 @@ int trace_enter_write(__u64 fd, char *buf, __u64 size) {
         req.protocol = PROTOCOL_MEMCACHED;
     } else if (is_mysql_query(buf, size)) {
         req.protocol = PROTOCOL_MYSQL;
+    } else if (is_mongo_query(buf, size)) {
+        req.protocol = PROTOCOL_MONGO;
     } else {
         return 0;
     }
@@ -144,6 +150,13 @@ int trace_exit_read(struct trace_event_raw_sys_exit_rw__stub* ctx) {
         e.status = parse_memcached_status(args->buf, ctx->ret);
     } else if (req->protocol == PROTOCOL_MYSQL) {
         e.status = parse_mysql_status(args->buf, ctx->ret);
+    } else if (req->protocol == PROTOCOL_MONGO) {
+        e.status = parse_mongo_status(args->buf, ctx->ret, req->partial);
+        if (e.status == 1) {
+            req->partial = 1;
+            bpf_map_update_elem(&active_l7_requests, &k, req, BPF_ANY);
+            return 0;
+        }
     }
     if (e.status == 0) {
         return 0;

ebpftracer/ebpf/l7.c

@@ -0,0 +1,55 @@
+// https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/
+
+#define MONGO_OP_COMPRESSED 2012
+#define MONGO_OP_MSG        2013
+
+struct mongo_header {
+    __s32 length;
+    __s32 request_id;
+    __s32 response_to;
+    __s32 op_code;
+};
+
+static __always_inline
+int is_mongo_query(char *buf, int buf_size) {
+    if (buf_size < 1) {
+        return 0;
+    }
+    struct mongo_header h = {};
+    if (bpf_probe_read(&h, sizeof(h), (void *)((char *)buf)) < 0) {
+        return 0;
+    }
+    if (h.response_to == 0 && (h.op_code == MONGO_OP_MSG || h.op_code == MONGO_OP_COMPRESSED)) {
+        return 1;
+    }
+    return 0;
+}
+
+static __always_inline
+__u32 parse_mongo_status(char *buf, int buf_size, __u8 partial) {
+    if (partial == 0 && buf_size == 4) { //partial read
+        return 1;
+    }
+    struct mongo_header h = {};
+    if (partial) {
+        if (bpf_probe_read(&h.response_to, sizeof(h.response_to), (void *)((char *)buf+4)) < 0) {
+            return 0;
+        }
+        if (bpf_probe_read(&h.op_code, sizeof(h.op_code), (void *)((char *)buf+8)) < 0) {
+            return 0;
+        }
+    } else {
+        if (bpf_probe_read(&h, sizeof(h), (void *)((char *)buf)) < 0) {
+            return 0;
+        }
+    }
+    if (h.response_to == 0) {
+        return 0;
+    }
+    if (h.op_code == MONGO_OP_MSG || h.op_code == MONGO_OP_COMPRESSED) {
+        return 200;
+    }
+    return 0;
+}
+
+

ebpftracer/ebpf/mongo.c

@@ -47,6 +47,7 @@ const (
 	L7ProtocolRedis     L7Protocol = 3
 	L7ProtocolMemcached L7Protocol = 4
 	L7ProtocolMysql     L7Protocol = 5
+	L7ProtocolMongo     L7Protocol = 6
 )
 
 func (p L7Protocol) String() string {
@@ -61,6 +62,8 @@ func (p L7Protocol) String() string {
 		return "Memcached"
 	case L7ProtocolMysql:
 		return "Mysql"
+	case L7ProtocolMongo:
+		return "Mongo"
 	}
 	return "UNKNOWN:" + strconv.Itoa(int(p))
 }