[sample #12602] [v2] feat(bpf): implement connection limit enforcement (QoS max connections)

felix/bpf-gpl/bpf.h

@@ -343,6 +343,8 @@ extern const volatile struct cali_tc_preamble_globals __globals;
 #define INGRESS_PACKET_RATE_CONFIGURED (GLOBAL_FLAGS & CALI_GLOBALS_INGRESS_PACKET_RATE_CONFIGURED)
 #define EGRESS_PACKET_RATE_CONFIGURED (GLOBAL_FLAGS & CALI_GLOBALS_EGRESS_PACKET_RATE_CONFIGURED)
 #define WORKLOAD_SRC_SPOOFING_CONFIGURED (GLOBAL_FLAGS & CALI_GLOBALS_WORKLOAD_SRC_SPOOFING_CONFIGURED)
+#define INGRESS_CONN_LIMIT_CONFIGURED (GLOBAL_FLAGS & CALI_GLOBALS_INGRESS_CONN_LIMIT_CONFIGURED)
+#define EGRESS_CONN_LIMIT_CONFIGURED (GLOBAL_FLAGS & CALI_GLOBALS_EGRESS_CONN_LIMIT_CONFIGURED)
 
 #define MAP_VERSIONED(name, ver) name##ver
 #define map_symbol(name, ver)  MAP_VERSIONED(name, ver)

felix/bpf-gpl/conntrack.h

@@ -11,6 +11,7 @@
 #include "icmp.h"
 #include "types.h"
 #include "rpf.h"
+#include "qos.h"
 
 #ifdef IPVER6
 #define IPPROTO_ICMP_46	IPPROTO_ICMPV6
@@ -574,6 +575,47 @@ static CALI_BPF_INLINE bool tcp_recycled(bool syn, struct calico_ct_value *v)
 	return syn && (a->fin_seen || a->rst_seen) && (b->fin_seen || b->rst_seen);
 }
 
+/* qos_connlimit_decrement_for_ct decrements the per-pod connlimit counter(s)
+ * for a CT entry that is closing or being purged. Reads the entry's
+ * CONNLIMIT_* flags and the per-leg ifindex fields to pick the right
+ * (ifindex, direction) pair(s). Idempotent: skips entries that already carry
+ * CONNLIMIT_DEC, and sets DEC before decrementing so a concurrent decrement
+ * on another CPU sees the flag and bails.
+ *
+ * Called from the packet path on FIN-FIN / RST close, and from
+ * conntrack_cleanup when an idle entry is purged with no close packet.
+ */
+static CALI_BPF_INLINE void qos_connlimit_decrement_for_ct(struct calico_ct_value *v)
+{
+	__u32 cl_flags = ct_value_get_flags(v);
+	if (cl_flags & CALI_CT_FLAG_CONNLIMIT_DEC) {
+		return;
+	}
+	if (!(cl_flags & (CALI_CT_FLAG_CONNLIMIT_INGRESS | CALI_CT_FLAG_CONNLIMIT_EGRESS))) {
+		return;
+	}
+	ct_value_set_flags(v, CALI_CT_FLAG_CONNLIMIT_DEC);
+
+	if (cl_flags & CALI_CT_FLAG_CONNLIMIT_INGRESS) {
+		/* Ingress: pod is the responder (non-opener). */
+		__u32 pod_ifindex = v->a_to_b.opener
+			? v->b_to_a.ifindex
+			: v->a_to_b.ifindex;
+		if (pod_ifindex != CT_INVALID_IFINDEX) {
+			qos_connlimit_decrement(pod_ifindex, 1);
+		}
+	}
+	if (cl_flags & CALI_CT_FLAG_CONNLIMIT_EGRESS) {
+		/* Egress: pod is the opener. */
+		__u32 pod_ifindex = v->a_to_b.opener
+			? v->a_to_b.ifindex
+			: v->b_to_a.ifindex;
+		if (pod_ifindex != CT_INVALID_IFINDEX) {
+			qos_connlimit_decrement(pod_ifindex, 0);
+		}
+	}
+}
+
 static CALI_BPF_INLINE struct calico_ct_result calico_ct_lookup(struct cali_tc_ctx *ctx)
 {
 	struct ct_lookup_ctx ct_lookup_ctx = {
@@ -997,6 +1039,14 @@ static CALI_BPF_INLINE struct calico_ct_result calico_ct_lookup(struct cali_tc_c
 			}
 		}
 		ct_tcp_entry_update(ctx, tcp_header, src_to_dst, dst_to_src);
+
+		/* Decrement connlimit counter when a TCP connection closes
+		 * (both FINs seen or RST). The helper sets CONNLIMIT_DEC
+		 * before decrementing so concurrent paths bail.
+		 */
+		if ((src_to_dst->fin_seen && dst_to_src->fin_seen) || tcp_header->rst) {
+			qos_connlimit_decrement_for_ct(v);
+		}
 	}
 
 	__u32 ifindex = CALI_F_TO_HOST ? ctx->skb->ifindex : skb_ingress_ifindex(ctx->skb);
@@ -1097,8 +1147,26 @@ static CALI_BPF_INLINE struct calico_ct_result calico_ct_lookup(struct cali_tc_c
 	if (syn) {
 		CALI_CT_DEBUG("packet is SYN");
 		ct_result_set_flag(result.rc, CT_RES_SYN);
-	}
 
+		/* For ingress connection limit: mark the CT entry when a SYN is
+		 * first seen at a WEP ingress. The CONNLIMIT_INGRESS flag on the
+		 * CT value prevents double-counting on SYN retransmissions.
+		 * CT_RES_CONNLIMIT_FIRST_SYN is set in the result ONLY on the
+		 * first SYN, so the accepted_entrypoint knows to increment.
+		 */
+		if (CALI_F_TO_WEP && INGRESS_CONN_LIMIT_CONFIGURED) {
+			if (!(ct_value_get_flags(v) & CALI_CT_FLAG_CONNLIMIT_INGRESS)) {
+				ct_value_set_flags(v, CALI_CT_FLAG_CONNLIMIT_INGRESS);
+				ct_result_set_flag(result.rc, CT_RES_CONNLIMIT_FIRST_SYN);
+			}
+			/* Propagate REJECTED flag so accepted_entrypoint can
+			 * unconditionally reject retransmissions of previously
+			 * rejected connections. */
+			if (ct_value_get_flags(v) & CALI_CT_FLAG_CONNLIMIT_REJECTED) {
+				result.flags |= CALI_CT_FLAG_CONNLIMIT_REJECTED;
+			}
+		}
+	}
 
 	CALI_CT_DEBUG("result: 0x%x", result.rc);
 

felix/bpf-gpl/conntrack_cleanup.c

@@ -41,6 +41,10 @@ static long process_ccq_entry(void *map, struct calico_ct_key *key, struct cali_
 	if (!value->rev_key.protocol) {
 		actual_ct_value = cali_ct_lookup_elem(key);
 		if (actual_ct_value && (actual_ct_value->last_seen == value->last_seen)) {
+			// Decrement the per-pod connlimit counter if this entry
+			// carried one of the CONNLIMIT_* flags and the packet
+			// path didn't already decrement.
+			qos_connlimit_decrement_for_ct(actual_ct_value);
 			if (!cali_ct_delete_elem(key)) {
 				ictx->num_cleaned++;
 			}
@@ -58,6 +62,8 @@ static long process_ccq_entry(void *map, struct calico_ct_key *key, struct cali_
 		}
 		struct calico_ct_value *rev_ct_value = cali_ct_lookup_elem(rev_key);
 		if (rev_ct_value && (rev_ct_value->last_seen == value->rev_last_seen)) {
+			// The reverse leg holds the connlimit flags + ifindex.
+			qos_connlimit_decrement_for_ct(rev_ct_value);
 			if (!cali_ct_delete_elem(rev_key)) {
 				ictx->num_cleaned++;
 			}

felix/bpf-gpl/conntrack_types.h

@@ -42,6 +42,10 @@ enum cali_ct_type {
 #define CALI_CT_FLAG_SET_DSCP	0x8000 /* marks connections that needs to set DSCP */
 #define CALI_CT_FLAG_MAGLEV	0X10000 /* marks Maglev connections. Allows packets of an existing to arrive via a different tunnel after failover. */
 #define CALI_CT_FLAG_SEND_RESET	0x20000 /* marks connections where we should send a TCP RST on behalf of the workload */
+#define CALI_CT_FLAG_CONNLIMIT_INGRESS	0x40000 /* marks connections counted against an ingress connection limit */
+#define CALI_CT_FLAG_CONNLIMIT_REJECTED	0x80000 /* marks connections rejected by the connection limit */
+#define CALI_CT_FLAG_CONNLIMIT_EGRESS	0x100000 /* marks connections counted against an egress connection limit */
+#define CALI_CT_FLAG_CONNLIMIT_DEC	0x200000 /* marks connections already decremented from connlimit counter */
 
 struct calico_ct_leg {
 	__u64 bytes;
@@ -134,6 +138,13 @@ static CALI_BPF_INLINE void __xxx_compile_asserts(void) {
 	ret;												\
 })
 
+#define ct_value_clear_flags(v, f) do {		\
+	(v)->flags  &= ~((f) & 0xff);		\
+	(v)->flags2 &= ~(((f) >> 8) & 0xff);	\
+	(v)->flags3 &= ~(((f) >> 16) & 0xff);	\
+	(v)->flags4 &= ~(((f) >> 24) & 0xff);	\
+} while(0)
+
 struct ct_lookup_ctx {
 	__u8 proto;
 	DECLARE_IP_ADDR(src);
@@ -219,6 +230,7 @@ enum calico_ct_result_type {
 #define CT_RES_SYN		0x1000
 #define CT_RES_CONFIRMED	0x2000
 #define CT_RES_TO_WORKLOAD	0x4000
+#define CT_RES_CONNLIMIT_FIRST_SYN	0x8000
 
 #define ct_result_rc(rc)			((rc) & 0xff)
 #define ct_result_flags(rc)			((rc) & ~0xff)

felix/bpf-gpl/globals.h

@@ -67,6 +67,8 @@ enum cali_globals_flags {
 	CALI_GLOBALS_EGRESS_PACKET_RATE_CONFIGURED        = 0x00008000,
 	CALI_GLOBALS_UDP_GSO_LINEARIZE                    = 0x00010000,
 	CALI_GLOBALS_WORKLOAD_SRC_SPOOFING_CONFIGURED     = 0x00020000,
+	CALI_GLOBALS_INGRESS_CONN_LIMIT_CONFIGURED        = 0x00040000,
+	CALI_GLOBALS_EGRESS_CONN_LIMIT_CONFIGURED         = 0x00080000,
 };
 
 struct cali_ctlb_globals {

felix/bpf-gpl/qos.h

@@ -17,17 +17,20 @@ struct calico_qos_key {
 
 struct calico_qos_val {
 	struct bpf_spin_lock lock;
-	// config
+	// packet rate config
 	__s16 packet_rate;
 	__s16 packet_burst;
-	// state
+	// packet rate state
 	__s16 packet_rate_tokens;
 	__s16 padding[3]; // alignment
 	__u64 packet_rate_last_update;
+	// connection limit
+	__s32 max_connections; // 0 = no limit, >0 = limit
+	__s32 current_count;   // maintained by BPF (increment on SYN) and scanner (recount)
 };
 
 // 2*IFACE_STATE_MAP_SIZE because it will potentially have 2 entries for each interface (ingress/egress)
-CALI_MAP(cali_qos,,
+CALI_MAP(cali_qos, 2,
 		BPF_MAP_TYPE_HASH,
 		struct calico_qos_key, struct calico_qos_val,
 		2*IFACE_STATE_MAP_SIZE, BPF_F_NO_PREALLOC)
@@ -151,4 +154,81 @@ static CALI_BPF_INLINE bool qos_dscp_set(struct cali_tc_ctx *ctx, __s8 dscp)
 	return true;
 }
 
+/* qos_connlimit_check_and_increment atomically checks the connection limit for
+ * the current interface and direction using the cali_qos map. If below the
+ * limit, increments the counter and returns 0 (allow). If at or above the
+ * limit, returns -1 (reject). Returns 0 if no limit is configured for this
+ * interface+direction (no map entry or max_connections <= 0).
+ *
+ * Only called for TCP SYN on WEP interfaces.
+ */
+static CALI_BPF_INLINE int qos_connlimit_check_and_increment(struct cali_tc_ctx *ctx)
+{
+	struct calico_qos_key key = {
+		.ifindex = ctx->skb->ifindex,
+#if CALI_F_INGRESS
+		.ingress = 1,
+#else // CALI_F_EGRESS
+		.ingress = 0,
+#endif
+	};
+
+	struct calico_qos_val *qos = cali_qos_lookup_elem(&key);
+	if (!qos || qos->max_connections <= 0) {
+		return 0;
+	}
+
+	int rc = 0;
+
+	bpf_spin_lock(&qos->lock);
+	if (qos->current_count >= qos->max_connections) {
+		rc = -1;
+	} else {
+		qos->current_count++;
+	}
+	bpf_spin_unlock(&qos->lock);
+
+	if (rc < 0) {
+		CALI_DEBUG("connlimit: over limit (%d >= %d), rejecting",
+			   qos->current_count, qos->max_connections);
+	} else {
+		CALI_DEBUG("connlimit: under limit (%d/%d), allowing",
+			   qos->current_count, qos->max_connections);
+	}
+
+	return rc;
+}
+
+/* qos_connlimit_decrement decrements the connection limit counter for the
+ * given interface and direction. Called when a TCP connection closes (both FINs
+ * seen or RST) and from the cleanup program when a counted CT entry expires.
+ * Takes an explicit ifindex+direction because the decrement may run from any
+ * BPF program (from_hep, from_wep, conntrack_cleanup, ...), not just the pod's
+ * own WEP program.
+ */
+static CALI_BPF_INLINE void qos_connlimit_decrement(__u32 ifindex, __u32 direction)
+{
+	struct calico_qos_key key = {
+		.ifindex = ifindex,
+		.ingress = direction,
+	};
+
+	struct calico_qos_val *qos = cali_qos_lookup_elem(&key);
+	if (!qos || qos->max_connections <= 0) {
+		return;
+	}
+
+	bpf_spin_lock(&qos->lock);
+	if (qos->current_count > 0) {
+		qos->current_count--;
+	}
+	bpf_spin_unlock(&qos->lock);
+
+	/* CALI_DEBUG_NO_FLAG (not CALI_DEBUG) — the TC programs' CALI_LOG
+	 * override reads ctx->globals which we don't have here.
+	 */
+	CALI_DEBUG_NO_FLAG("connlimit: decremented count to %d/%d",
+		   qos->current_count, qos->max_connections);
+}
+
 #endif /* __CALI_QOS_H__ */

felix/bpf-gpl/tc.c

@@ -1401,6 +1401,52 @@ int calico_tc_skb_accepted_entrypoint(struct __sk_buff *skb)
 		skb_log(ctx, true);
 	}
 
+	/* Ingress connection limit for TCP SYN arriving at a WEP.
+	 * On the first SYN (CT_RES_CONNLIMIT_FIRST_SYN), atomically check the
+	 * limit and increment the counter in the QoS map. On SYN retransmissions,
+	 * always reject if the CT entry was previously rejected (CONNLIMIT_REJECTED
+	 * flag), otherwise allow (the connection was already accepted and counted).
+	 * The Go-side CT scanner periodically recounts and corrects drift.
+	 */
+	if (CALI_F_TO_WEP && !policy_skipped &&
+			ctx->state->ip_proto == IPPROTO_TCP &&
+			ct_result_is_syn(ctx->state->ct_result.rc)) {
+		bool reject = false;
+		if (ctx->state->ct_result.rc & CT_RES_CONNLIMIT_FIRST_SYN) {
+			/* First SYN: check and increment. */
+			reject = qos_connlimit_check_and_increment(ctx) < 0;
+		} else if (ctx->state->ct_result.flags & CALI_CT_FLAG_CONNLIMIT_REJECTED) {
+			/* SYN retransmission of a previously rejected connection. */
+			CALI_DEBUG("connlimit: retransmission of rejected connection");
+			reject = true;
+		}
+		if (reject) {
+			CALI_DEBUG("Ingress connection limit exceeded, rejecting with TCP RST");
+			/* Mark the CT entry so retransmissions are unconditionally
+			 * rejected without re-checking the counter (which may have
+			 * changed due to scanner recounts). */
+			{
+				bool sltd = src_lt_dest(&ctx->state->ip_src,
+							&ctx->state->ip_dst,
+							ctx->state->sport,
+							ctx->state->dport);
+				struct calico_ct_key ck;
+				fill_ct_key(&ck, sltd, ctx->state->ip_proto,
+					    &ctx->state->ip_src, &ctx->state->ip_dst,
+					    ctx->state->sport, ctx->state->dport);
+				struct calico_ct_value *cv = cali_ct_lookup_elem(&ck);
+				if (cv) {
+					ct_value_set_flags(cv, CALI_CT_FLAG_CONNLIMIT_REJECTED);
+				}
+			}
+			if (!skb_refresh_validate_ptrs(ctx, TCP_SIZE)) {
+				ctx->state->ct_result.ifindex_fwd = CT_INVALID_IFINDEX;
+				CALI_JUMP_TO(ctx, PROG_INDEX_TCP_RST);
+			}
+			goto deny;
+		}
+	}
+
 	if ((CALI_F_FROM_WEP || CALI_F_TO_HEP) && qos_dscp_needs_update(ctx) && !qos_dscp_set(ctx, EGRESS_DSCP)) {
 		goto deny;
 	}
@@ -1475,6 +1521,22 @@ int calico_tc_skb_new_flow_entrypoint(struct __sk_buff *skb)
 		CALI_DEBUG("Allowed by policy: ACCEPT");
 	}
 
+	/* Check egress connection limit for new TCP connections from WEP.
+	 * Atomically check the limit and increment the counter in the QoS map.
+	 * The Go-side CT scanner periodically recounts and corrects drift.
+	 */
+	if (CALI_F_FROM_WEP && state->ip_proto == IPPROTO_TCP &&
+			!(state->flags & CALI_ST_SUPPRESS_CT_STATE)) {
+		if (qos_connlimit_check_and_increment(ctx) < 0) {
+			CALI_DEBUG("Egress connection limit exceeded, rejecting with TCP RST");
+			if (!skb_refresh_validate_ptrs(ctx, TCP_SIZE)) {
+				ctx->state->ct_result.ifindex_fwd = CT_INVALID_IFINDEX;
+				CALI_JUMP_TO(ctx, PROG_INDEX_TCP_RST);
+			}
+			goto deny;
+		}
+	}
+
 	if (CALI_F_FROM_WEP &&
 			CALI_DROP_WORKLOAD_TO_HOST &&
 			cali_rt_flags_local_host(
@@ -1512,6 +1574,9 @@ int calico_tc_skb_new_flow_entrypoint(struct __sk_buff *skb)
 	if (state->flags & CALI_ST_SKIP_REDIR_PEER) {
 		ct_ctx_nat->flags |= CALI_CT_FLAG_SKIP_REDIR_PEER;
 	}
+	if (CALI_F_FROM_WEP && state->ip_proto == IPPROTO_TCP && EGRESS_CONN_LIMIT_CONFIGURED) {
+		ct_ctx_nat->flags |= CALI_CT_FLAG_CONNLIMIT_EGRESS;
+	}
 	if (CALI_F_TO_WEP) {
 		if (!(ctx->skb->mark & CALI_SKB_MARK_SEEN)) {
 			/* If the packet wasn't seen, must come from host. There is no