Taehwan Dev Blog

쿠버네티스, eBPF, 그리고 머신러닝 이야기

View My GitHub Profile

20 March 2026

eBPF로 Pod Blocker 구현

by Yang Taehwan

프로젝트 소개

깃헙 레포지토리

개발 스택

개발환경 구축

시나리오

쿠버네티스 네트워크 구성

eBPF DaemonSet 구성

apiVersion: apps/v1
kind: DaemonSet
# ...
    spec:
# ...
      hostNetwork: true
      hostPID: true
      nodeSelector:
        kubernetes.io/os: linux
      tolerations:
        - operator: Exists
      containers:
# ...
          securityContext:
            privileged: true
# ...
          volumeMounts:
            - name: bpffs
              mountPath: /sys/fs/bpf
            - name: debugfs
              mountPath: /sys/kernel/debug
            - name: lib-modules
              mountPath: /lib/modules
              readOnly: true
            - name: host-proc
              mountPath: /host/proc
              readOnly: true
      volumes:
        - name: bpffs
          hostPath:
            path: /sys/fs/bpf
            type: DirectoryOrCreate
        - name: debugfs
          hostPath:
            path: /sys/kernel/debug
            type: DirectoryOrCreate
        - name: lib-modules
          hostPath:
            path: /lib/modules
        - name: host-proc
          hostPath:
            path: /proc

구현 상세

auth-attacker

func main() {
  // ...
  transport := &http.Transport{
		DisableKeepAlives: true,
	}

	client := &http.Client{
		Timeout:   5 * time.Second,
		Transport: transport,
	}

  var wg sync.WaitGroup
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go worker(&wg, client, jobs, &success, &failure)
	}
  // ...
}

func worker(
  wg *sync.WaitGroup,
	client *http.Client,
	jobs <-chan int,
	success *int64,
	failure *int64,
) {
  for jobID := range jobs {
		// ...
		err := sendRequest(client)
    // ...
  }
}

func sendRequest(client *http.Client) error {
  // ...  
	req, err := http.NewRequest(http.MethodPost, tokenURL, bytes.NewBufferString(form.Encode()))
  // ...
	resp, err := client.Do(req)
	// ...
}

eBPF 커널모드

// target_ip가 보호하고자 하는 파드의 IP이다.
struct ip_pair_key {
    __u32 target_ip;
    __u32 src_ip;
};

// window_ns는 요청을 카운트하는 시간
// max_count는 최대 가능 요청 횟수
// 예를 들어, src 파드에서 target 파드로 1분에 100번 커넥션 요청이 들어오면 패킷을 드롭한다는 정책이 된다.
struct rl_config {
    __u64 window_ns;
    __u64 max_count;
    __u32 pad; 
};

// 설정을 저장할 자료구조
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 1);
    __type(key, __u32);
    __type(value, struct rl_config);
} config_map SEC(".maps");

// 유저모드에서 전달받은 보호할 파드 목록
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 4096);
    __type(key, __u32);
    __type(value, __u8);
} watch_dst_ips SEC(".maps");

struct rl_state {
    struct bpf_spin_lock lock;
    __u64 window_start_ns;
    __u32 count;
    __u32 pad;
};

struct drop_event {
    __u64 ts_ns;
    __u32 target_ip;
    __u32 src_ip;
    __u32 count;
    __u32 max_count;
};

// (target_ip, src_ip)별 rate limit 상태 저장
// key = {target_ip, src_ip}
// value = rl_state
// 요청 횟수를 저장하기 위한 자료구조
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 65536);
    __type(key, struct ip_pair_key);
    __type(value, struct rl_state);
} target_src_states SEC(".maps");

// 패킷 드롭 이벤트를 유저모드로 전달하기 위한 링 버퍼
struct {
    __uint(type, BPF_MAP_TYPE_RINGBUF);
    __uint(max_entries, 1 << 24); // 16MiB
} drop_events SEC(".maps");

SEC("tc")
int count_syn_and_drop(struct __sk_buff *skb) 
{
    // ...
    //  패킷을 IP 헤더와 TCP 해더로 파싱
    if (parse_ipv4_tcp(skb, &ip, &tcp) < 0) {
        return TC_ACT_OK;
    }
    // ...
    // 요청 횟수를 기록 중인 맵을 가져온다
    state = bpf_map_lookup_elem(&target_src_states, &pair_key);
    // ...
    // 리밋을 넘지 않았다면 횟수에 1을 더한다.
    state->count += 1;
    // 요청 횟수가 최대 가능 요청 횟수를 초과하면 패킷을 드롭한다.
    if (state->count > cfg->max_count) {
        __u32 current_count = state->count;
        __u32 max_count = cfg->max_count;
        __u32 target_ip = pair_key.target_ip;
        __u32 src_ip = pair_key.src_ip;

        bpf_spin_unlock(&state->lock);

        bpf_printk("drop target=%x src=%x count=%u\n",
               target_ip, src_ip, current_count);

        emit_drop_event(now_ns, target_ip, src_ip, current_count, max_count);
        return TC_ACT_SHOT;
    }
    // ...    
    return TC_ACT_OK;
}

eBPF 유저모드

type Agent struct {
	objs     count_conn_and_dropObjects
	tcClient *tc.Tc
	watchSet map[uint32]struct{}
	ifIndex  uint32
	ifName   string
	tcHandle uint32
}

func CreateTCHookAndShowDropLog(ctx context.Context) error {
  // ...
  // app=authorization-server 레이블을 가진 파드를 가져온다.
  pods := PodsByLabel()
  
  if err := agent.applyRateLimitConfig(Window, MaxCount); err != nil {
    // ...
  }
  // ...
  if err := agent.setWatchIPs(podIPs); err != nil {
    // ...
  }
  // ...
}

kubectl exec auth-test-authorization-server-7ddc9bcc8d-kqd87 -- cat /sys/class/net/eth0/iflink
12
ip link
# veth54c1b554가 authorization server 파드의 host-side veth이다.
12: veth54c1b554@if2: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1450 qdisc noqueue master cni0 state UP mode DEFAULT group default qlen 1000
    link/ether 42:33:3d:19:04:6d brd ff:ff:ff:ff:ff:ff link-netns cni-4f918ed8-3dcf-7039-ca38-3602426ed6cc

func CreateTCHookAndShowDropLog(ctx context.Context) error {
  // ...
  hostVeth, err := hostVethFromPod(pods)
  // ...
  iface, err := net.InterfaceByName(hostVeth)
  // ...
  tcClient, err := tc.Open(&tc.Config{})
  // ...
  if err := resetClsact(agent.tcClient, agent.ifIndex); err != nil {
    // ...
  }
  // ...
  if err := attachBPFProgram(
		agent.tcClient,
		agent.ifIndex,
		agent.objs.CountSynAndDrop.FD(),
		"count_syn_and_drop",
		agent.tcHandle,
	); err != nil {
    // ...
  }
  // ...
}

func attachBPFProgram(tcnl *tc.Tc, ifindex uint32, progFD int, progName string, handle uint32) error {
	fd := uint32(progFD)
	flags := uint32(0x1) // direct-action

	filter := tc.Object{
		Msg: tc.Msg{
			Family:  unix.AF_UNSPEC,
      // authorization server의 host-side veth
			Ifindex: ifindex,
			Handle:  handle,
      // tc hook을 host-side veth의 tc egress로 걸어야 한다.
			Parent:  core.BuildHandle(tc.HandleRoot, tc.HandleMinEgress),
			Info:    core.FilterInfo(0, unix.ETH_P_ALL),
		},
		Attribute: tc.Attribute{
			Kind: "bpf",
			BPF: &tc.Bpf{
				FD:    &fd,
				Name:  &progName,
				Flags: &flags,
			},
		},
	}
	return tcnl.Filter().Add(&filter)
}

func CreateTCHookAndShowDropLog(ctx context.Context) error {
  // ...
  return runDropEventLoop(ctx, reader)
}

func runDropEventLoop(ctx context.Context, reader *ringbuf.Reader) error {
  for {
		record, err := reader.Read()
    // ...
    var evt dropEvent
		if err := binary.Read(bytes.NewReader(record.RawSample), binary.LittleEndian, &evt); err != nil {
      // ...
    }

    log.Printf(
			"[DROP] recv_time=%s src=%s dst=%s count=%d limit=%d kernel_ts_ns=%d",
			time.Now().Format(time.RFC3339),
			u32ToIP(evt.SrcIp),
			u32ToIP(evt.TargetIp),
			evt.Count,
			evt.MaxCount,
			evt.TsNs,
		)
  }
}
tags: