parse mysql prepared stmt

Author: kenanfarukcakir

aggregator/data.go

@@ -78,6 +78,10 @@ type Aggregator struct {
 	pgStmtsMu sync.RWMutex
 	pgStmts   map[string]string // pid-fd-stmtname -> query
 
+	// postgres prepared stmt
+	mySqlStmtsMu sync.RWMutex
+	mySqlStmts   map[string]string // pid-fd-stmtId -> query
+
 	liveProcessesMu sync.RWMutex
 	liveProcesses   map[uint32]struct{} // pid -> struct{}
 
@@ -153,6 +157,7 @@ func NewAggregator(parentCtx context.Context, ct *cri.CRITool, k8sChan chan inte
 		liveProcesses:       make(map[uint32]struct{}),
 		rateLimiters:        make(map[uint32]*rate.Limiter),
 		pgStmts:             make(map[string]string),
+		mySqlStmts:          make(map[string]string),
 	}
 
 	var err error
@@ -381,6 +386,14 @@ func (a *Aggregator) processExit(pid uint32) {
 		}
 	}
 	a.pgStmtsMu.Unlock()
+
+	a.mySqlStmtsMu.Lock()
+	for key, _ := range a.pgStmts {
+		if strings.HasPrefix(key, fmt.Sprint(pid)) {
+			delete(a.mySqlStmts, key)
+		}
+	}
+	a.mySqlStmtsMu.Unlock()
 }
 
 func (a *Aggregator) signalTlsAttachment(pid uint32) {
@@ -1373,17 +1386,29 @@ func (a *Aggregator) findRelatedSocket(ctx context.Context, d *l7_req.L7Event) (
 func (a *Aggregator) parseMySQLCommand(d *l7_req.L7Event) (string, error) {
 	r := d.Payload[:d.PayloadSize]
 	var sqlCommand string
+	// 3 bytes len, 1 byte package number, 1 byte command type
+	if len(r) < 5 {
+		return "", fmt.Errorf("too short for a sql query")
+	}
+	r = r[5:]
+	sqlCommand = string(r)
 	if d.Method == l7_req.MYSQL_TEXT_QUERY {
-		// 3 bytes len, 1 byte package number, 1 byte command type
-		if len(r) < 5 {
-			return "", fmt.Errorf("too short for a sql query")
-		}
-		r = r[5:]
-		sqlCommand = string(r)
-
 		if !containsSQLKeywords(sqlCommand) {
 			return "", fmt.Errorf("no sql command found")
 		}
+	} else if d.Method == l7_req.MYSQL_PREPARE_STMT {
+		a.mySqlStmtsMu.Lock()
+		a.mySqlStmts[fmt.Sprintf("%d-%d-%d", d.Pid, d.Fd, d.MySqlPrepStmtId)] = string(r)
+		a.mySqlStmtsMu.Unlock()
+	} else if d.Method == l7_req.MYSQL_EXEC_STMT {
+		a.mySqlStmtsMu.RLock()
+		query, ok := a.mySqlStmts[fmt.Sprintf("%d-%d-%d", d.Pid, d.Fd, d.MySqlPrepStmtId)]
+		a.mySqlStmtsMu.RUnlock()
+		if !ok || query == "" { // we don't have the query for the prepared statement
+			// Execute (name of prepared statement) [(parameter)]
+			return fmt.Sprintf("EXECUTE %d *values*", d.MySqlPrepStmtId), nil
+		}
+		sqlCommand = query
 	}
 	return sqlCommand, nil
 }

ebpf/c/bpf_bpfeb.go

@@ -31,10 +31,12 @@ struct l7_event {
     __u8 is_tls;
 
     __u32 seq; // tcp sequence number
-    __u32 tid;
+    __u32 tid; // thread id
 
     __s16 kafka_api_version; // used only for kafka
+    __u32 prep_statement_id; // used only for mysql
 
+    // socket pair
     __u32 saddr;
     __u16 sport;
     __u32 daddr;
@@ -816,7 +818,7 @@ int process_exit_of_syscalls_read_recvfrom(void* ctx, __u64 id, __u32 pid, __s64
                 e->kafka_api_version = active_req->api_version;
             }
         }else if (e->protocol == PROTOCOL_MYSQL) {
-            e->status = is_mysql_response(read_info->buf, ret);
+            e->status = is_mysql_response(read_info->buf, ret, active_req->request_type, &(e->prep_statement_id));
             e->method = METHOD_UNKNOWN;
             if (active_req->request_type == MYSQL_COM_STMT_PREPARE) {
                 e->method = METHOD_MYSQL_PREPARE_STMT;

ebpf/c/bpf_bpfeb.o

@@ -25,8 +25,8 @@
 #define METHOD_MYSQL_PREPARE_STMT 2
 #define METHOD_MYSQL_EXEC_STMT 3
 
-#define MYSQL_STATUS_FAILED 0
 #define MYSQL_STATUS_OK 1
+#define MYSQL_STATUS_FAILED 2
 
 static __always_inline
 int is_mysql_query(char *buf, __u64 buf_size, __u8 *request_type) {
@@ -63,8 +63,8 @@ int is_mysql_query(char *buf, __u64 buf_size, __u8 *request_type) {
 
 // __u32 *statement_id
 static __always_inline
-int is_mysql_response(char *buf, __u64 buf_size) {
-    __u8 b[5]; // first 5 bytes, first 3 represents length, 4th is packet number, 5th is command type
+int is_mysql_response(char *buf, __u64 buf_size, __u8 request_type, __u32 *statement_id) {
+    __u8 b[5]; // first 5 bytes, first 3 represents length, 4th is packet number, 5th is response code
     if (bpf_probe_read(&b, sizeof(b), (void *)((char *)buf)) < 0) {
         return 0;
     }
@@ -77,6 +77,12 @@ int is_mysql_response(char *buf, __u64 buf_size) {
         return MYSQL_STATUS_OK;
     }
     if (b[4] == MYSQL_RESPONSE_OK) {
+        if (request_type == MYSQL_COM_STMT_PREPARE) {
+            // 6-9th bytes returns statement id
+            if (bpf_probe_read(statement_id, sizeof(*statement_id), (void *)((char *)buf+5)) < 0) {
+                return 0;
+            }
+        }
         return MYSQL_STATUS_OK;
     }
     if (b[4] == MYSQL_RESPONSE_ERROR) {

ebpf/c/bpf_bpfel.go

@@ -353,12 +353,13 @@ type bpfL7Event struct {
 	Tid                 uint32
 	KafkaApiVersion     int16
 	_                   [2]byte
+	PrepStatementId     uint32 // for mysql
 	Saddr               uint32
 	Sport               uint16
 	_                   [2]byte
 	Daddr               uint32
 	Dport               uint16
-	_                   [2]byte
+	_                   [6]byte
 }
 
 type bpfTraceEvent struct {
@@ -401,6 +402,7 @@ type L7Event struct {
 	Tid                 uint32
 	Seq                 uint32 // tcp seq num
 	KafkaApiVersion     int16
+	MySqlPrepStmtId     uint32
 	Saddr               uint32
 	Sport               uint16
 	Daddr               uint32
@@ -739,6 +741,7 @@ func (l7p *L7Prog) Consume(ctx context.Context, ch chan interface{}) {
 				Tid:                 l7Event.Tid,
 				Seq:                 l7Event.Seq,
 				KafkaApiVersion:     l7Event.KafkaApiVersion,
+				MySqlPrepStmtId:     l7Event.PrepStatementId,
 				Saddr:               l7Event.Saddr,
 				Sport:               l7Event.Sport,
 				Daddr:               l7Event.Daddr,
@@ -750,6 +753,8 @@ func (l7p *L7Prog) Consume(ctx context.Context, ch chan interface{}) {
 					Str("payload", string(userspacel7Event.Payload[:userspacel7Event.PayloadSize])).
 					Uint32("pid", userspacel7Event.Pid).
 					Str("method", userspacel7Event.Method).
+					Uint32("stmtId", userspacel7Event.MySqlPrepStmtId).
+					Uint64("fd", userspacel7Event.Fd).
 					Msg("mysql-event")
 			}