parse sql host address and port

.gitignore

@@ -85,3 +85,4 @@ test/integration/components/go_otel_grpc/rolldice
 test/integration/components/pythonserver/lib/
 test/integration/components/pythonserver/lib64
 test/integration/components/pythonserver/pyvenv.cfg
+test/integration/components/gosql/qosql

bpf/go_common.h

@@ -74,6 +74,41 @@ struct {
  __uint(pinning, BEYLA_PIN_INTERNAL);
 } go_trace_map SEC(".maps");
 
+struct {
+ __uint(type, BPF_MAP_TYPE_LRU_HASH);
+ __type(key, go_addr_key_t); // key: goroutine
+ __type(value, void *); // the transport *
+ __uint(max_entries, MAX_CONCURRENT_REQUESTS);
+} ongoing_grpc_operate_headers SEC(".maps");
+
+typedef struct grpc_transports {
+ u8 type;
+ connection_info_t conn;
+} grpc_transports_t;
+
+// TODO: use go_addr_key_t as key
+struct {
+ __uint(type, BPF_MAP_TYPE_LRU_HASH);
+ __type(key, void *); // key: pointer to the transport pointer
+ __type(value, grpc_transports_t);
+ __uint(max_entries, MAX_CONCURRENT_REQUESTS);
+} ongoing_grpc_transports SEC(".maps");
+
+typedef struct sql_func_invocation {
+ u64 start_monotime_ns;
+ u64 sql_param;
+ u64 query_len;
+ connection_info_t conn __attribute__((aligned(8)));
+ tp_info_t tp;
+} sql_func_invocation_t;
+
+struct {
+ __uint(type, BPF_MAP_TYPE_LRU_HASH);
+ __type(key, go_addr_key_t); // key: pointer to the request goroutine
+ __type(value, sql_func_invocation_t);
+ __uint(max_entries, MAX_CONCURRENT_REQUESTS);
+} ongoing_sql_queries SEC(".maps");
+
 static __always_inline void go_addr_key_from_id(go_addr_key_t *current, void *addr) {
  u64 pid_tid = bpf_get_current_pid_tgid();
  u32 pid = pid_from_pid_tgid(pid_tid);

bpf/go_grpc.h

@@ -42,26 +42,6 @@ typedef struct grpc_client_func_invocation {
  u64 flags;
 } grpc_client_func_invocation_t;
 
-typedef struct grpc_transports {
- u8 type;
- connection_info_t conn;
-} grpc_transports_t;
-
-// TODO: use go_addr_key_t as key
-struct {
- __uint(type, BPF_MAP_TYPE_LRU_HASH);
- __type(key, void *); // key: pointer to the transport pointer
- __type(value, grpc_transports_t);
- __uint(max_entries, MAX_CONCURRENT_REQUESTS);
-} ongoing_grpc_transports SEC(".maps");
-
-struct {
- __uint(type, BPF_MAP_TYPE_LRU_HASH);
- __type(key, go_addr_key_t); // key: goroutine
- __type(value, void *); // the transport *
- __uint(max_entries, MAX_CONCURRENT_REQUESTS);
-} ongoing_grpc_operate_headers SEC(".maps");
-
 struct {
  __uint(type, BPF_MAP_TYPE_LRU_HASH);
  __type(key, go_addr_key_t); // key: pointer to the request goroutine
@@ -141,28 +121,6 @@ int uprobe_server_handleStream(struct pt_regs *ctx) {
  return 0;
 }
 
-// Sets up the connection info to be grabbed and mapped over the transport to operateHeaders
-SEC("uprobe/netFdReadGRPC")
-int uprobe_netFdReadGRPC(struct pt_regs *ctx) {
- void *goroutine_addr = GOROUTINE_PTR(ctx);
- bpf_dbg_printk("=== uprobe/proc netFD read goroutine %lx === ", goroutine_addr);
- go_addr_key_t g_key = {};
- go_addr_key_from_id(&g_key, goroutine_addr);
-
- void *tr = bpf_map_lookup_elem(&ongoing_grpc_operate_headers, &g_key);
- bpf_dbg_printk("tr %llx", tr);
- if (tr) {
- grpc_transports_t *t = bpf_map_lookup_elem(&ongoing_grpc_transports, tr);
- bpf_dbg_printk("t %llx", t);
- if (t) {
- void *fd_ptr = GO_PARAM1(ctx);
- get_conn_info_from_fd(fd_ptr, &t->conn); // ok to not check the result, we leave it as 0
- }
- }
-
- return 0;
-}
-
 // Handles finding the connection information for http2 servers in grpc
 SEC("uprobe/http2Server_operateHeaders")
 int uprobe_http2Server_operateHeaders(struct pt_regs *ctx) {

bpf/go_nethttp.h

@@ -912,18 +912,42 @@ int uprobe_netFdRead(struct pt_regs *ctx) {
  go_addr_key_t g_key = {};
  go_addr_key_from_id(&g_key, goroutine_addr);
 
+ // lookup active HTTP connection
  connection_info_t *conn = bpf_map_lookup_elem(&ongoing_server_connections, &g_key);
-
  if (conn) {
- bpf_dbg_printk(
- "Found existing server connection, parsing FD information for socket tuples, %llx",
- goroutine_addr);
-
- void *fd_ptr = GO_PARAM1(ctx);
- get_conn_info_from_fd(fd_ptr, conn); // ok to not check the result, we leave it as 0
+ if (conn->d_port == 0 && conn->s_port == 0) {
+ bpf_dbg_printk(
+ "Found existing server connection, parsing FD information for socket tuples, %llx",
+ goroutine_addr);
 
+ void *fd_ptr = GO_PARAM1(ctx);
+ get_conn_info_from_fd(fd_ptr, conn); // ok to not check the result, we leave it as 0
+ }
  //dbg_print_http_connection_info(conn);
  }
+ // lookup a grpc connection
+ // Sets up the connection info to be grabbed and mapped over the transport to operateHeaders
+ void *tr = bpf_map_lookup_elem(&ongoing_grpc_operate_headers, &g_key);
+ bpf_dbg_printk("tr %llx", tr);
+ if (tr) {
+ grpc_transports_t *t = bpf_map_lookup_elem(&ongoing_grpc_transports, tr);
+ bpf_dbg_printk("t %llx", t);
+ if (t) {
+ if (t->conn.d_port == 0 && t->conn.s_port == 0) {
+ void *fd_ptr = GO_PARAM1(ctx);
+ get_conn_info_from_fd(fd_ptr,
+ &t->conn); // ok to not check the result, we leave it as 0
+ }
+ }
+ }
+ // lookup active sql connection
+ sql_func_invocation_t *sql_conn = bpf_map_lookup_elem(&ongoing_sql_queries, &g_key);
+ bpf_dbg_printk("sql_conn %llx", sql_conn);
+ if (sql_conn) {
+ void *fd_ptr = GO_PARAM1(ctx);
+ get_conn_info_from_fd(fd_ptr,
+ &sql_conn->conn); // ok to not check the result, we leave it as 0
+ }
 
  return 0;
 }

bpf/go_sql.h

@@ -26,24 +26,11 @@
 #include "go_common.h"
 #include "ringbuf.h"
 
-typedef struct sql_func_invocation {
- u64 start_monotime_ns;
- u64 sql_param;
- u64 query_len;
- tp_info_t tp;
-} sql_func_invocation_t;
-
-struct {
- __uint(type, BPF_MAP_TYPE_LRU_HASH);
- __type(key, go_addr_key_t); // key: pointer to the request goroutine
- __type(value, sql_func_invocation_t);
- __uint(max_entries, MAX_CONCURRENT_REQUESTS);
-} ongoing_sql_queries SEC(".maps");
-
 static __always_inline void set_sql_info(void *goroutine_addr, void *sql_param, void *query_len) {
  sql_func_invocation_t invocation = {.start_monotime_ns = bpf_ktime_get_ns(),
  .sql_param = (u64)sql_param,
  .query_len = (u64)query_len,
+ .conn = {0},
  .tp = {0}};
 
  // We don't look up in the headers, no http/grpc request, therefore 0 as last argument
@@ -65,6 +52,7 @@ int uprobe_queryDC(struct pt_regs *ctx) {
 
  void *sql_param = GO_PARAM8(ctx);
  void *query_len = GO_PARAM9(ctx);
+
  set_sql_info(goroutine_addr, sql_param, query_len);
  return 0;
 }
@@ -117,6 +105,9 @@ int uprobe_queryReturn(struct pt_regs *ctx) {
  if (query_len < sizeof(trace->sql)) {
  trace->sql[query_len] = 0;
  }
+
+ __builtin_memcpy(&trace->conn, &invocation->conn, sizeof(connection_info_t));
+
  // submit the completed trace via ringbuffer
  bpf_ringbuf_submit(trace, get_flags());
  } else {

bpf/tracer_common.h

@@ -51,6 +51,7 @@ typedef struct sql_request_trace_t {
  u64 end_monotime_ns;
  u8 sql[SQL_MAX_LEN];
  u16 status;
+ connection_info_t conn __attribute__((aligned(8)));
  tp_info_t tp;
  pid_info pid;
 } __attribute__((packed)) sql_request_trace;

pkg/internal/ebpf/common/spanner.go

@@ -76,14 +76,25 @@ func SQLRequestTraceToSpan(trace *SQLRequestTrace) request.Span {
 
  method, path := sqlprune.SQLParseOperationAndTable(sql)
 
+ peer := ""
+ peerPort := 0
+ hostname := ""
+ hostPort := 0
+
+ if trace.Conn.S_port != 0 || trace.Conn.D_port != 0 {
+ peer, hostname = (*BPFConnInfo)(unsafe.Pointer(&trace.Conn)).reqHostInfo()
+ peerPort = int(trace.Conn.S_port)
+ hostPort = int(trace.Conn.D_port)
+ }
+
  return request.Span{
  Type: request.EventType(trace.Type),
  Method: method,
  Path: path,
- Peer: "",
- PeerPort: 0,
- Host: "",
- HostPort: 0,
+ Peer: peer,
+ PeerPort: peerPort,
+ Host: hostname,
+ HostPort: hostPort,
  ContentLength: 0,
  RequestStart: int64(trace.StartMonotimeNs),
  Start: int64(trace.StartMonotimeNs),

pkg/internal/ebpf/common/sql_detect_transform.go

@@ -209,11 +209,13 @@ func parsePosgresQueryCommand(buf []byte) (string, error) {
 
 func TCPToSQLToSpan(trace *TCPRequestInfo, op, table, sql string) request.Span {
  peer := ""
+ peerPort := 0
  hostname := ""
  hostPort := 0
 
  if trace.ConnInfo.S_port != 0 || trace.ConnInfo.D_port != 0 {
  peer, hostname = (*BPFConnInfo)(unsafe.Pointer(&trace.ConnInfo)).reqHostInfo()
+ peerPort = int(trace.ConnInfo.S_port)
  hostPort = int(trace.ConnInfo.D_port)
  }
 
@@ -222,7 +224,7 @@ func TCPToSQLToSpan(trace *TCPRequestInfo, op, table, sql string) request.Span {
  Method: op,
  Path: table,
  Peer: peer,
- PeerPort: int(trace.ConnInfo.S_port),
+ PeerPort: peerPort,
  Host: hostname,
  HostPort: hostPort,
  ContentLength: 0,