Report TCP Failed Connections

go.mod

@@ -604,7 +604,7 @@ require github.com/lorenzosaino/go-sysctl v0.3.1
 
 require (
  github.com/DATA-DOG/go-sqlmock v1.5.2
- github.com/DataDog/agent-payload/v5 v5.0.114
+ github.com/DataDog/agent-payload/v5 v5.0.116-0.20240515152545-ac55806c4516
  github.com/DataDog/datadog-agent/cmd/agent/common/path v0.54.0-rc.2
  github.com/DataDog/datadog-agent/comp/core/config v0.54.0-rc.2
  github.com/DataDog/datadog-agent/comp/core/flare/types v0.54.0-rc.2

pkg/config/setup/system_probe.go

@@ -195,6 +195,7 @@ func InitSystemProbeConfig(cfg pkgconfigmodel.Config) {
  cfg.BindEnvAndSetDefault(join(spNS, "enable_conntrack_all_namespaces"), true, "DD_SYSTEM_PROBE_ENABLE_CONNTRACK_ALL_NAMESPACES")
  cfg.BindEnvAndSetDefault(join(netNS, "enable_protocol_classification"), true, "DD_ENABLE_PROTOCOL_CLASSIFICATION")
  cfg.BindEnvAndSetDefault(join(netNS, "enable_ringbuffers"), true, "DD_SYSTEM_PROBE_NETWORK_ENABLE_RINGBUFFERS")
+ cfg.BindEnvAndSetDefault(join(netNS, "enable_tcp_failed_connections"), false, "DD_SYSTEM_PROBE_NETWORK_ENABLE_TCP_FAILED_CONNS")
  cfg.BindEnvAndSetDefault(join(netNS, "ignore_conntrack_init_failure"), false, "DD_SYSTEM_PROBE_NETWORK_IGNORE_CONNTRACK_INIT_FAILURE")
  cfg.BindEnvAndSetDefault(join(netNS, "conntrack_init_timeout"), 10*time.Second)
  cfg.BindEnvAndSetDefault(join(netNS, "allow_netlink_conntracker_fallback"), true)

pkg/network/config/config.go

@@ -260,6 +260,9 @@ type Config struct {
  // classifying the L7 protocols being used.
  ProtocolClassificationEnabled bool
 
+ // TCPFailedConnectionsEnabled specifies whether the tracer will track & report TCP error codes
+ TCPFailedConnectionsEnabled bool
+
  // EnableHTTPStatsByStatusCode specifies if the HTTP stats should be aggregated by the actual status code
  // instead of the status code family.
  EnableHTTPStatsByStatusCode bool
@@ -312,6 +315,7 @@ func New() *Config {
  ExcludedSourceConnections: cfg.GetStringMapStringSlice(join(spNS, "source_excludes")),
  ExcludedDestinationConnections: cfg.GetStringMapStringSlice(join(spNS, "dest_excludes")),
 
+ TCPFailedConnectionsEnabled: cfg.GetBool(join(netNS, "enable_failed_connections")),
  MaxTrackedConnections: uint32(cfg.GetInt64(join(spNS, "max_tracked_connections"))),
  MaxClosedConnectionsBuffered: uint32(cfg.GetInt64(join(spNS, "max_closed_connections_buffered"))),
  ClosedConnectionFlushThreshold: cfg.GetInt(join(spNS, "closed_connection_flush_threshold")),

pkg/network/ebpf/c/tracer.c

@@ -18,6 +18,12 @@
 #include "tracer/tcp_recv.h"
 #include "protocols/classification/protocol-classification.h"
 
+__maybe_unused static __always_inline bool tcp_failed_connections_enabled() {
+ __u64 val = 0;
+ LOAD_CONSTANT("tcp_failed_connections_enabled", val);
+ return val > 0;
+}
+
 SEC("socket/classifier_entry")
 int socket__classifier_entry(struct __sk_buff *skb) {
  protocol_classifier_entrypoint(skb);
@@ -215,11 +221,33 @@ int kprobe__tcp_close(struct pt_regs *ctx) {
  return 0;
 }
 
+SEC("kprobe/tcp_done")
+int kprobe__tcp_done(struct pt_regs *ctx) {
+ struct sock *sk;
+ conn_tuple_t t = {};
+ u64 pid_tgid = bpf_get_current_pid_tgid();
+ sk = (struct sock *)PT_REGS_PARM1(ctx);
+
+ log_debug("kprobe/tcp_done: tgid: %llu, pid: %llu", pid_tgid >> 32, pid_tgid & 0xFFFFFFFF);
+ if (!read_conn_tuple(&t, sk, pid_tgid, CONN_TYPE_TCP)) {
+ return 0;
+ }
+ log_debug("kprobe/tcp_done: netns: %u, sport: %u, dport: %u", t.netns, t.sport, t.dport);
+
+ int err = 0;
+ bpf_probe_read_kernel_with_telemetry(&err, sizeof(err), (&sk->sk_err));
+ if (err != 0 && tcp_failed_connections_enabled()) {
+ flush_tcp_failure(ctx, &t, err);
+ return 0;
+ }
+ return 0;
+}
+
 SEC("kretprobe/tcp_close")
 int kretprobe__tcp_close_clean_protocols(struct pt_regs *ctx) {
  u64 pid_tgid = bpf_get_current_pid_tgid();
 
- conn_tuple_t *tup_ptr = (conn_tuple_t*) bpf_map_lookup_elem(&tcp_close_args, &pid_tgid);
+ conn_tuple_t *tup_ptr = (conn_tuple_t *)bpf_map_lookup_elem(&tcp_close_args, &pid_tgid);
  if (tup_ptr) {
  clean_protocol_classification(tup_ptr);
  bpf_map_delete_elem(&tcp_close_args, &pid_tgid);
@@ -440,7 +468,6 @@ int kretprobe__ip6_make_skb(struct pt_regs *ctx) {
 
 #endif // !COMPILE_RUNTIME || FEATURE_UDPV6_ENABLED
 
-
 static __always_inline u32 fl4_saddr(struct flowi4 *fl4) {
  u32 addr = 0;
 #ifdef COMPILE_PREBUILT
@@ -548,7 +575,7 @@ int kprobe__ip_make_skb(struct pt_regs *ctx) {
  struct flowi4 *fl4 = (struct flowi4 *)PT_REGS_PARM2(ctx);
 #if defined(COMPILE_PREBUILT) || defined(COMPILE_CORE) || (defined(COMPILE_RUNTIME) && LINUX_VERSION_CODE >= KERNEL_VERSION(4, 18, 0))
  unsigned int flags = PT_REGS_PARM10(ctx);
- if (flags&MSG_SPLICE_PAGES && udp_send_page_enabled()) {
+ if (flags & MSG_SPLICE_PAGES && udp_send_page_enabled()) {
  return 0;
  }
 #endif
@@ -600,18 +627,18 @@ int kretprobe__ip_make_skb(struct pt_regs *ctx) {
  return handle_ip_skb(sk, size, fl4);
 }
 
-#define handle_udp_recvmsg(sk, msg, flags, udp_sock_map) \
- do { \
- log_debug("kprobe/udp_recvmsg: flags: %x", flags); \
- if (flags & MSG_PEEK) { \
- return 0; \
- } \
- \
+#define handle_udp_recvmsg(sk, msg, flags, udp_sock_map)  \
+ do {  \
+ log_debug("kprobe/udp_recvmsg: flags: %x", flags);  \
+ if (flags & MSG_PEEK) {  \
+ return 0;  \
+ }  \
+  \
  /* keep track of non-peeking calls, since skb_free_datagram_locked doesn't have that argument */ \
- u64 pid_tgid = bpf_get_current_pid_tgid(); \
- udp_recv_sock_t t = { .sk = sk, .msg = msg }; \
- bpf_map_update_with_telemetry(udp_sock_map, &pid_tgid, &t, BPF_ANY); \
- return 0; \
+ u64 pid_tgid = bpf_get_current_pid_tgid();  \
+ udp_recv_sock_t t = { .sk = sk, .msg = msg };  \
+ bpf_map_update_with_telemetry(udp_sock_map, &pid_tgid, &t, BPF_ANY);  \
+ return 0;  \
  } while (0);
 
 SEC("kprobe/udp_recvmsg")
@@ -807,7 +834,6 @@ int kprobe__skb_consume_udp(struct pt_regs *ctx) {
  return handle_skb_consume_udp(sk, skb, len);
 }
 
-
 #ifdef COMPILE_PREBUILT
 
 SEC("kprobe/tcp_retransmit_skb")
@@ -882,12 +908,12 @@ int kretprobe__tcp_retransmit_skb(struct pt_regs *ctx) {
  if (args == NULL) {
  return 0;
  }
- struct sock* sk = args->sk;
+ struct sock *sk = args->sk;
  u32 retrans_out_pre = args->retrans_out_pre;
  bpf_map_delete_elem(&pending_tcp_retransmit_skb, &tid);
  u32 retrans_out = 0;
  BPF_CORE_READ_INTO(&retrans_out, tcp_sk(sk), retrans_out);
- return handle_retransmit(sk, retrans_out-retrans_out_pre);
+ return handle_retransmit(sk, retrans_out - retrans_out_pre);
 }
 
 #endif // COMPILE_CORE || COMPILE_RUNTIME
@@ -1054,13 +1080,13 @@ int kretprobe__inet6_bind(struct pt_regs *ctx) {
 // Represents the parameters being passed to the tracepoint net/net_dev_queue
 struct net_dev_queue_ctx {
  u64 unused;
- struct sk_buff* skb;
+ struct sk_buff *skb;
 };
 
-static __always_inline struct sock* sk_buff_sk(struct sk_buff *skb) {
- struct sock * sk = NULL;
+static __always_inline struct sock *sk_buff_sk(struct sk_buff *skb) {
+ struct sock *sk = NULL;
 #ifdef COMPILE_PREBUILT
- bpf_probe_read(&sk, sizeof(struct sock*), (char*)skb + offset_sk_buff_sock());
+ bpf_probe_read(&sk, sizeof(struct sock *), (char *)skb + offset_sk_buff_sock());
 #elif defined(COMPILE_CORE) || defined(COMPILE_RUNTIME)
  BPF_CORE_READ_INTO(&sk, skb, sk);
 #endif
@@ -1069,12 +1095,12 @@ static __always_inline struct sock* sk_buff_sk(struct sk_buff *skb) {
 }
 
 SEC("tracepoint/net/net_dev_queue")
-int tracepoint__net__net_dev_queue(struct net_dev_queue_ctx* ctx) {
- struct sk_buff* skb = ctx->skb;
+int tracepoint__net__net_dev_queue(struct net_dev_queue_ctx *ctx) {
+ struct sk_buff *skb = ctx->skb;
  if (!skb) {
  return 0;
  }
- struct sock* sk = sk_buff_sk(skb);
+ struct sock *sk = sk_buff_sk(skb);
  if (!sk) {
  return 0;
  }
@@ -1085,7 +1111,7 @@ int tracepoint__net__net_dev_queue(struct net_dev_queue_ctx* ctx) {
  return 0;
  }
 
- if (!(skb_tup.metadata&CONN_TYPE_TCP)) {
+ if (!(skb_tup.metadata & CONN_TYPE_TCP)) {
  return 0;
  }
 

pkg/network/ebpf/c/tracer/events.h

@@ -34,7 +34,7 @@ static __always_inline void clean_protocol_classification(conn_tuple_t *tup) {
  bpf_map_delete_elem(&conn_tuple_to_socket_skb_conn_tuple, &conn_tuple);
 }
 
-__maybe_unused static __always_inline void submit_event(void *ctx, int cpu, void *event_data, size_t data_size) {
+__maybe_unused static __always_inline void submit_closed_conn_event(void *ctx, int cpu, void *event_data, size_t data_size) {
  __u64 ringbuffers_enabled = 0;
  LOAD_CONSTANT("ringbuffers_enabled", ringbuffers_enabled);
  if (ringbuffers_enabled > 0) {
@@ -128,7 +128,7 @@ static __always_inline void cleanup_conn(void *ctx, conn_tuple_t *tup, struct so
  // We send the connection outside of a batch anyway. This is likely not as
  // frequent of a case to cause performance issues and avoid cases where
  // we drop whole connections, which impacts things USM connection matching.
- submit_event(ctx, cpu, &conn, sizeof(conn_t));
+ submit_closed_conn_event(ctx, cpu, &conn, sizeof(conn_t));
  if (is_tcp) {
  increment_telemetry_count(unbatched_tcp_close);
  }
@@ -137,8 +137,22 @@ static __always_inline void cleanup_conn(void *ctx, conn_tuple_t *tup, struct so
  }
 }
 
+// This function is used to flush the conn_failed_t to the perf or ring buffer.
+static __always_inline void flush_tcp_failure(void *ctx, conn_tuple_t *tup, int failure_reason) {
+ conn_failed_t failure = {};
+ failure.tup = *tup;
+ failure.failure_reason = failure_reason;
+
+ __u64 ringbuffers_enabled = 0;
+ LOAD_CONSTANT("ringbuffers_enabled", ringbuffers_enabled);
+ if (ringbuffers_enabled > 0) {
+ bpf_ringbuf_output(&failed_conn_event, &failure, sizeof(conn_failed_t), 0);
+ } else {
+ u32 cpu = bpf_get_smp_processor_id();
+ bpf_perf_event_output(ctx, &conn_close_event, cpu, &failure, sizeof(conn_failed_t));
+ }
+}
 
-// This function is used to flush the conn_close_batch to the perf or ring buffer.
 static __always_inline void flush_conn_close_if_full(void *ctx) {
  u32 cpu = bpf_get_smp_processor_id();
  batch_t *batch_ptr = bpf_map_lookup_elem(&conn_close_batch, &cpu);
@@ -154,7 +168,7 @@ static __always_inline void flush_conn_close_if_full(void *ctx) {
  batch_ptr->len = 0;
  batch_ptr->id++;
 
- submit_event(ctx, cpu, &batch_copy, sizeof(batch_t));
+ submit_closed_conn_event(ctx, cpu, &batch_copy, sizeof(batch_t));
 }
 
 #endif // __TRACER_EVENTS_H

pkg/network/ebpf/c/tracer/maps.h

@@ -31,6 +31,9 @@ BPF_HASH_MAP(tcp_ongoing_connect_pid, struct sock *, __u64, 1024)
  */
 BPF_PERF_EVENT_ARRAY_MAP(conn_close_event, __u32)
 
+/* Will hold TCP failed connections
+ */
+BPF_PERF_EVENT_ARRAY_MAP(failed_conn_event, __u32)
 /* We use this map as a container for batching closed tcp/udp connections
  * The key represents the CPU core. Ideally we should use a BPF_MAP_TYPE_PERCPU_HASH map
  * or BPF_MAP_TYPE_PERCPU_ARRAY, but they are not available in

pkg/network/ebpf/c/tracer/tracer.h

@@ -10,15 +10,13 @@
 #define true 1
 #define false 0
 
-typedef enum
-{
+typedef enum {
  CONN_DIRECTION_UNKNOWN = 0b00,
  CONN_DIRECTION_INCOMING = 0b01,
  CONN_DIRECTION_OUTGOING = 0b10,
 } conn_direction_t;
 
-typedef enum
-{
+typedef enum {
  PACKET_COUNT_NONE = 0,
  PACKET_COUNT_ABSOLUTE = 1,
  PACKET_COUNT_INCREMENT = 2,
@@ -54,8 +52,7 @@ typedef struct {
 } conn_stats_ts_t;
 
 // Connection flags
-typedef enum
-{
+typedef enum {
  CONN_L_INIT = 1 << 0, // initial/first message sent
  CONN_R_INIT = 1 << 1, // reply received for initial message from remote
  CONN_ASSURED = 1 << 2 // "3-way handshake" complete, i.e. response to initial reply sent
@@ -77,6 +74,11 @@ typedef struct {
  __u32 tcp_retransmits;
 } conn_t;
 
+typedef struct {
+ conn_tuple_t tup;
+ __u32 failure_reason;
+} conn_failed_t;
+
 // Must match the number of conn_t objects embedded in the batch_t struct
 #ifndef CONN_CLOSED_BATCH_SIZE
 #define CONN_CLOSED_BATCH_SIZE 4

pkg/network/ebpf/kprobe.go

@@ -58,6 +58,24 @@ func (t ConnTuple) DestEndpoint() string {
  return net.JoinHostPort(t.DestAddress().String(), strconv.Itoa(int(t.Dport)))
 }
 
+// SetFamily sets the family (IPv4 or IPv6) for a tuple.
+func (t *ConnTuple) SetFamily(family ConnFamily) {
+ if family == IPv6 {
+ t.Metadata |= uint32(IPv6) // Set the IPv6 bit
+ } else {
+ t.Metadata &^= uint32(IPv6) // Clear the IPv6 bit
+ }
+}
+
+// SetType sets the type (TCP or UDP) for a tuple.
+func (t *ConnTuple) SetType(connType ConnType) {
+ if connType == TCP {
+ t.Metadata |= uint32(TCP) // Set the TCP bit
+ } else {
+ t.Metadata &^= uint32(TCP) // Clear the TCP bit
+ }
+}
+
 func (t ConnTuple) String() string {
  return fmt.Sprintf(
  "[%s%s] [PID: %d] [%s ⇄ %s] (ns: %d)",

pkg/network/ebpf/kprobe_types.go

@@ -20,6 +20,7 @@ type ConnTuple C.conn_tuple_t
 type TCPStats C.tcp_stats_t
 type ConnStats C.conn_stats_ts_t
 type Conn C.conn_t
+type FailedConn C.conn_failed_t
 type Batch C.batch_t
 type Telemetry C.telemetry_t
 type PortBinding C.port_binding_t
@@ -53,6 +54,7 @@ const BatchSize = C.CONN_CLOSED_BATCH_SIZE
 const SizeofBatch = C.sizeof_batch_t
 
 const SizeofConn = C.sizeof_conn_t
+const SizeofFailedConn = C.sizeof_conn_failed_t
 
 type ClassificationProgram = uint32
 

pkg/network/ebpf/probes/probes.go

@@ -82,6 +82,8 @@ const (
 
  // TCPClose traces the tcp_close() system call
  TCPClose ProbeFuncName = "kprobe__tcp_close"
+ // TCPDone traces the tcp_done() system call
+ TCPDone ProbeFuncName = "kprobe__tcp_done"
  // TCPCloseCleanProtocolsReturn traces the return of tcp_close() system call
  TCPCloseCleanProtocolsReturn ProbeFuncName = "kretprobe__tcp_close_clean_protocols"
  // TCPCloseFlushReturn traces the return of tcp_close() system call
@@ -192,6 +194,8 @@ const (
  TCPConnectSockPidMap BPFMapName = "tcp_ongoing_connect_pid"
  // ConnCloseEventMap is the map storing connection close events
  ConnCloseEventMap BPFMapName = "conn_close_event"
+ // FailedConnEventMap is the map for storing failed connection events
+ FailedConnEventMap BPFMapName = "failed_conn_event"
  // TracerStatusMap is the map storing the status of the tracer
  TracerStatusMap BPFMapName = "tracer_status"
  // ConntrackStatusMap is the map storing the status of the conntrack