Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Change Go to use net.FD for peer and host #725

Merged
merged 8 commits into from
Apr 11, 2024
Merged

Change Go to use net.FD for peer and host #725

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d751acc
Use net.FD everywhere for Go connection info
grcevski Apr 8, 2024
b6b080c
remove unused fields
grcevski Apr 8, 2024
d018786
More remove
grcevski Apr 8, 2024
e82ddd1
Fix unit test
grcevski Apr 8, 2024
980b0d8
Fix map sizing to allow for proper working on larger request rates
grcevski Apr 9, 2024
8f61784
Merge branch 'main' into change_go_ip_report
grcevski Apr 9, 2024
7cb6f41
update offsets
grcevski Apr 9, 2024
35600c8
Fix test, we use ips now
grcevski Apr 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 67 additions & 3 deletions bpf/go_common.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,12 @@
#include "trace_util.h"
#include "go_traceparent.h"

volatile const u64 conn_fd_pos;
volatile const u64 fd_laddr_pos;
volatile const u64 fd_raddr_pos;
volatile const u64 tcp_addr_port_ptr_pos;
volatile const u64 tcp_addr_ip_ptr_pos;

char __license[] SEC("license") = "Dual MIT/GPL";

// Temporary information about a function invocation. It stores the invocation time of a function
Expand Down Expand Up @@ -49,7 +55,14 @@ struct {
__type(key, void *); // key: pointer to the request goroutine
__type(value, connection_info_t);
__uint(max_entries, MAX_CONCURRENT_REQUESTS);
} ongoing_http_server_connections SEC(".maps");
} ongoing_server_connections SEC(".maps");

struct {
__uint(type, BPF_MAP_TYPE_LRU_HASH);
__type(key, void *); // key: pointer to the request goroutine
__type(value, connection_info_t);
__uint(max_entries, MAX_CONCURRENT_REQUESTS);
} ongoing_client_connections SEC(".maps");

struct {
__uint(type, BPF_MAP_TYPE_LRU_HASH);
Expand Down Expand Up @@ -126,7 +139,7 @@ static __always_inline void server_trace_parent(void *goroutine_addr, tp_info_t
decode_go_traceparent(buf, tp->trace_id, tp->parent_id, &tp->flags);
}
} else {
connection_info_t *info = bpf_map_lookup_elem(&ongoing_http_server_connections, &goroutine_addr);
connection_info_t *info = bpf_map_lookup_elem(&ongoing_server_connections, &goroutine_addr);
u8 found_info = 0;

if (info) {
Expand Down Expand Up @@ -195,5 +208,56 @@ static __always_inline u8 client_trace_parent(void *goroutine_addr, tp_info_t *t
return found_trace_id;
}

static __always_inline void read_ip_and_port(u8 *dst_ip, u16 *dst_port, void *src) {
s64 addr_len = 0;
void *addr_ip = 0;

bpf_probe_read(dst_port, sizeof(u16), (void *)(src + tcp_addr_port_ptr_pos));
bpf_probe_read(&addr_ip, sizeof(addr_ip), (void *)(src + tcp_addr_ip_ptr_pos));
if (addr_ip) {
bpf_probe_read(&addr_len, sizeof(addr_len), (void *)(src + tcp_addr_ip_ptr_pos + 8));
if (addr_len == 4) {
__builtin_memcpy(dst_ip, ip4ip6_prefix, sizeof(ip4ip6_prefix));
bpf_probe_read(dst_ip + sizeof(ip4ip6_prefix), 4, addr_ip);
} else if (addr_len == 16) {
bpf_probe_read(dst_ip, 16, addr_ip);
}
}
}

static __always_inline void get_conn_info_from_fd(void *fd_ptr, connection_info_t *info) {
if (fd_ptr) {
void *laddr_ptr = 0;
void *raddr_ptr = 0;

bpf_probe_read(&laddr_ptr, sizeof(laddr_ptr), (void *)(fd_ptr + fd_laddr_pos + 8)); // find laddr
bpf_probe_read(&raddr_ptr, sizeof(raddr_ptr), (void *)(fd_ptr + fd_raddr_pos + 8)); // find raddr

bpf_dbg_printk("laddr_ptr %llx, laddr %llx, raddr %llx", fd_ptr + fd_laddr_pos + 8, laddr_ptr, raddr_ptr);
if (laddr_ptr && raddr_ptr) {

// read local
read_ip_and_port(info->s_addr, &info->s_port, laddr_ptr);

// read remote
read_ip_and_port(info->d_addr, &info->d_port, raddr_ptr);

sort_connection_info(info);
//dbg_print_http_connection_info(info);
}
}
}

// HTTP black-box context propagation
static __always_inline void get_conn_info(void *conn_ptr, connection_info_t *info) {
if (conn_ptr) {
void *fd_ptr = 0;
bpf_probe_read(&fd_ptr, sizeof(fd_ptr), (void *)(conn_ptr + conn_fd_pos)); // find fd

bpf_dbg_printk("Found fd ptr %llx", fd_ptr);

get_conn_info_from_fd(fd_ptr, info);
}
}

#endif // GO_COMMON_H
#endif // GO_COMMON_H
106 changes: 40 additions & 66 deletions bpf/go_grpc.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,16 +78,12 @@ volatile const u64 grpc_stream_st_ptr_pos;
volatile const u64 grpc_stream_method_ptr_pos;
volatile const u64 grpc_status_s_pos;
volatile const u64 grpc_status_code_ptr_pos;
volatile const u64 grpc_st_remoteaddr_ptr_pos;
volatile const u64 grpc_st_localaddr_ptr_pos;
volatile const u64 grpc_st_peer_ptr_pos;
volatile const u64 tcp_addr_port_ptr_pos;
volatile const u64 tcp_addr_ip_ptr_pos;
volatile const u64 grpc_client_target_ptr_pos;
volatile const u64 grpc_stream_ctx_ptr_pos;
volatile const u64 value_context_val_ptr_pos;
volatile const u64 grpc_peer_addr_pos;
volatile const u64 grpc_peer_localaddr_pos;
volatile const u64 grpc_st_conn_pos;
volatile const u64 grpc_t_conn_pos;

// Context propagation
volatile const u64 http2_client_next_id_pos;
Expand Down Expand Up @@ -177,65 +173,30 @@ int uprobe_server_handleStream_return(struct pt_regs *ctx) {
}

void *st_ptr = 0;
u8 found_conn = 0;
// Read the embedded object ptr
bpf_probe_read(&st_ptr, sizeof(st_ptr), (void *)(stream_ptr + grpc_stream_st_ptr_pos + sizeof(void *)));

trace->host_len = 0;
trace->remote_addr_len = 0;
trace->host_port = 0;
__builtin_memset(trace->host, 0, sizeof(trace->host));
__builtin_memset(trace->remote_addr, 0, sizeof(trace->remote_addr));

bpf_dbg_printk("st_ptr %llx", st_ptr);
if (st_ptr) {
void *remote_addr = 0;
void *local_addr = 0;

if (grpc_st_peer_ptr_pos != (u64)(-1)) {
bpf_dbg_printk("1.60 grpc or later peer info");
remote_addr = (void *)(st_ptr + grpc_st_peer_ptr_pos + grpc_peer_addr_pos + sizeof(void *));
local_addr = (void *)(st_ptr + grpc_st_peer_ptr_pos + grpc_peer_localaddr_pos + sizeof(void *));
} else if (grpc_st_remoteaddr_ptr_pos != (u64)(-1)) {
bpf_dbg_printk("pre 1.60 grpc peer info");
remote_addr = (void *)(st_ptr + grpc_st_remoteaddr_ptr_pos + sizeof(void *));
local_addr = (void *)(st_ptr + grpc_st_localaddr_ptr_pos + sizeof(void *));
}

if (remote_addr != 0 && local_addr != 0) {
void *peer_ptr = 0;
bpf_probe_read(&peer_ptr, sizeof(peer_ptr), remote_addr);

if (peer_ptr) {
u64 remote_addr_len = 0;
if (!read_go_byte_arr("grpc peer ptr", peer_ptr, tcp_addr_ip_ptr_pos, &trace->remote_addr, &remote_addr_len, sizeof(trace->remote_addr))) {
bpf_printk("can't read grpc peer ptr");
bpf_ringbuf_discard(trace, 0);
goto done;
}
trace->remote_addr_len = remote_addr_len;
void *conn_ptr = st_ptr + grpc_st_conn_pos;
bpf_dbg_printk("conn_ptr %llx", conn_ptr);
if (conn_ptr) {
void *conn_conn_ptr = 0;
bpf_probe_read(&conn_conn_ptr, sizeof(conn_conn_ptr), conn_ptr + 8);
bpf_dbg_printk("conn_conn_ptr %llx", conn_conn_ptr);
if (conn_conn_ptr) {
get_conn_info(conn_conn_ptr, &trace->conn);
found_conn = 1;
}
}
}

void *host_ptr = 0;
bpf_probe_read(&host_ptr, sizeof(host_ptr), local_addr);

if (host_ptr) {
u64 host_len = 0;

if (!read_go_byte_arr("grpc host ptr", host_ptr, tcp_addr_ip_ptr_pos, &trace->host, &host_len, sizeof(trace->host))) {
bpf_printk("can't read grpc host ptr");
bpf_ringbuf_discard(trace, 0);
goto done;
}
trace->host_len = host_len;

bpf_probe_read(&trace->host_port, sizeof(trace->host_port), (void *)(host_ptr + tcp_addr_port_ptr_pos));
}
} else {
bpf_dbg_printk("can't find peer grpc information");
}
if (!found_conn) {
__builtin_memset(&trace->conn, 0, sizeof(connection_info_t));
}

trace->tp = invocation->tp;

trace->end_monotime_ns = bpf_ktime_get_ns();
// submit the completed trace via ringbuffer
bpf_ringbuf_submit(trace, get_flags());
Expand Down Expand Up @@ -380,7 +341,6 @@ int uprobe_ClientConn_Invoke_return(struct pt_regs *ctx) {
// Read arguments from the original set of registers

// Get client request value pointers
void *cc_ptr = (void *)invocation->cc;
void *method_ptr = (void *)invocation->method;
void *method_len = (void *)invocation->method_len;
void *err = (void *)GO_PARAM1(ctx);
Expand All @@ -394,11 +354,12 @@ int uprobe_ClientConn_Invoke_return(struct pt_regs *ctx) {
goto done;
}

// Get the host information of the remote
if (!read_go_str("host", cc_ptr, grpc_client_target_ptr_pos, &trace->host, sizeof(trace->host))) {
bpf_printk("can't read http Request.Host");
bpf_ringbuf_discard(trace, 0);
goto done;
connection_info_t *info = bpf_map_lookup_elem(&ongoing_client_connections, &goroutine_addr);

if (info) {
__builtin_memcpy(&trace->conn, info, sizeof(connection_info_t));
} else {
__builtin_memset(&trace->conn, 0, sizeof(connection_info_t));
}

trace->tp = invocation->tp;
Expand All @@ -417,15 +378,28 @@ int uprobe_ClientConn_Invoke_return(struct pt_regs *ctx) {
// We extract the stream ID when it's just created and make a mapping of it to our goroutine that's executing ClientConn.Invoke.
SEC("uprobe/transport_http2Client_NewStream")
int uprobe_transport_http2Client_NewStream(struct pt_regs *ctx) {
#ifndef NO_HEADER_PROPAGATION
bpf_dbg_printk("=== uprobe/proc transport.(*http2Client).NewStream === ");

void *goroutine_addr = GOROUTINE_PTR(ctx);
void *t_ptr = GO_PARAM1(ctx);

bpf_dbg_printk("goroutine_addr %lx, t_ptr %llx", goroutine_addr, t_ptr);
bpf_dbg_printk("goroutine_addr %lx, t_ptr %llx, t.conn_pos %x", goroutine_addr, t_ptr, grpc_t_conn_pos);

if (t_ptr) {
void *conn_ptr = t_ptr + grpc_t_conn_pos;
bpf_dbg_printk("conn_ptr %llx", conn_ptr);
if (conn_ptr) {
void *conn_conn_ptr = 0;
bpf_probe_read(&conn_conn_ptr, sizeof(conn_conn_ptr), conn_ptr + 8);
bpf_dbg_printk("conn_conn_ptr %llx", conn_conn_ptr);
if (conn_conn_ptr) {
connection_info_t conn = {0};
get_conn_info(conn_conn_ptr, &conn);
bpf_map_update_elem(&ongoing_client_connections, &goroutine_addr, &conn, BPF_ANY);
}
}

#ifndef NO_HEADER_PROPAGATION
u32 next_id = 0;
// Read the next stream id from the httpClient
bpf_probe_read(&next_id, sizeof(next_id), (void *)(t_ptr + http2_client_next_id_pos));
Expand All @@ -440,11 +414,11 @@ int uprobe_transport_http2Client_NewStream(struct pt_regs *ctx) {
// seen later by writeHeader to clean up this mapping.
bpf_map_update_elem(&ongoing_streams, &next_id, &inv_save, BPF_ANY);
} else {
bpf_dbg_printk("Couldn't find invocation metadata for goroutite %lx", goroutine_addr);
bpf_dbg_printk("Couldn't find invocation metadata for goroutine %lx", goroutine_addr);
}
#endif
}

#endif
return 0;
}

Expand Down
Loading
Loading