pwru 是 Cilium 推出的基于 eBPF 開(kāi)發(fā)的網(wǎng)絡(luò)數(shù)據(jù)包排查工具，它提供了更細(xì)粒度的網(wǎng)絡(luò)數(shù)據(jù)包排查方案。本文將介紹 pwru 的使用方法和經(jīng)典場(chǎng)景，并介紹其實(shí)現(xiàn)原理。

安裝部署

部署要求

pwru 要求內(nèi)核代碼在 5.5 版本之上，--output-skb 要求內(nèi)核版本在 5.9 之上，并且要求內(nèi)核開(kāi)啟以下配置：

使用方法

Usageof./pwru:
--filter-dst-ipstringfilterdestinationIPaddr
--filter-dst-portuint16filterdestinationport
--filter-funcstringfilterkernelfunctionstobeprobedbyname(exactmatch,supportsRE2regularexpression)
--filter-markuint32filterskbmark
--filter-netnsuint32filternetnsinode
--filter-protostringfilterL4protocol(tcp,udp,icmp)
--filter-src-ipstringfiltersourceIPaddr
--filter-src-portuint16filtersourceport
--output-limit-linesuintexittheprogramafterthenumberofeventshasbeenreceived/printed
--output-metaprintskbmetadata
--output-relative-timestampprintrelativetimestampperskb
--output-skbprintskb
--output-stackprintstack
--output-tupleprintL4tuple

案例演示

下圖案例演示了 pwru 展現(xiàn)出快速定位出數(shù)據(jù)包被 iptables 規(guī)則 drop 掉的原因：

在不設(shè)置 iptables 規(guī)則之前：

添加了 iptables 規(guī)則之后

iptables-tfilter-IOUTPUT1-mtcp--prototcp--dst1.1.1.1/32-jDROP

可以看到在 nf_hook_slow 函數(shù)后發(fā)生了變化：

我們可以看到數(shù)據(jù)包在 nf_hook_slow 判決為 NF_DROP，調(diào)用了 kfree_skb

intnf_hook_slow(structsk_buff*skb,structnf_hook_state*state,
conststructnf_hook_entries*e,unsignedints)
{
unsignedintverdict;
intret;

for(;snum_hook_entries;s++){
verdict=nf_hook_entry_hookfn(&e->hooks[s],skb,state);
switch(verdict&NF_VERDICT_MASK){
caseNF_ACCEPT:
break;
caseNF_DROP:
kfree_skb(skb);
ret=NF_DROP_GETERR(verdict);
if(ret==0)
ret=-EPERM;
returnret;
caseNF_QUEUE:
ret=nf_queue(skb,state,s,verdict);
if(ret==1)
continue;
returnret;
default:
/*ImplicithandlingforNF_STOLEN,aswellasanyother
*nonconventionalverdicts.
*/
return0;
}
}

return1;
}

原理實(shí)現(xiàn)

pwru 本質(zhì)上是向 kprobe 注冊(cè)了一些 eBPF code，根據(jù) pwru 傳入的參數(shù)可以更新 eBPF Map，改變限制條件，從而更新輸出。

比如在 FilterCfg 里面制定了過(guò)濾的 IP 地址和協(xié)議等條件

typeFilterCfgstruct{
FilterMarkuint32

//Filterl3
FilterIPv6uint8
FilterSrcIP[16]byte
FilterDstIP[16]byte

//Filterl4
FilterProtouint8
FilterSrcPortuint16
FilterDstPortuint16

//TODO:iftherearemoreoptionslater,thenyoucanconsiderusingabitmap
OutputRelativeTSuint8
OutputMetauint8
OutputTupleuint8
OutputSkbuint8
OutputStackuint8

Padbyte
}

會(huì)根據(jù) pwru 傳入的參數(shù)更新這個(gè) eBPF Map

funcConfigBPFMap(flags*Flags,cfgMap*ebpf.Map){
cfg:=FilterCfg{
FilterMark:flags.FilterMark,
}

ifflags.FilterSrcPort>0{
cfg.FilterSrcPort=byteorder.HostToNetwork16(flags.FilterSrcPort)
}
ifflags.FilterDstPort>0{
cfg.FilterDstPort=byteorder.HostToNetwork16(flags.FilterDstPort)
}


switchstrings.ToLower(flags.FilterProto){
case"tcp":
cfg.FilterProto=syscall.IPPROTO_TCP
case"udp":
cfg.FilterProto=syscall.IPPROTO_UDP
case"icmp":
cfg.FilterProto=syscall.IPPROTO_ICMP
case"icmp6":
cfg.FilterProto=syscall.IPPROTO_ICMPV6
}

//...

iferr:=cfgMap.Update(uint32(0),cfg,0);err!=nil{
log.Fatalf("Failedtosetfiltermap:%v",err)
}
}

在 eBPF code 中，可以看到會(huì)讀取配置 bpf_map_lookup_elem，然后進(jìn)而執(zhí)行真正的 filter：

structconfig{
u32mark;
u8ipv6;
unionaddrsaddr;
unionaddrdaddr;
u8l4_proto;
u16sport;
u16dport;
u8output_timestamp;
u8output_meta;
u8output_tuple;
u8output_skb;
u8output_stack;
u8pad;
}__attribute__((packed));

static__always_inlineint
handle_everything(structsk_buff*skb,structpt_regs*ctx){
structevent_tevent={};

u32index=0;
structconfig*cfg=bpf_map_lookup_elem(&cfg_map,&index);

if(cfg){
if(!filter(skb,cfg))
return0;

set_output(ctx,skb,&event,cfg);
}

event.pid=bpf_get_current_pid_tgid();
event.addr=PT_REGS_IP(ctx);
event.skb_addr=(u64)skb;
event.ts=bpf_ktime_get_ns();
bpf_perf_event_output(ctx,&events,BPF_F_CURRENT_CPU,&event,sizeof(event));

return0;
}

可以看到，這里通過(guò) bpf_perf_event_output 將過(guò)濾結(jié)果以 Perf event 傳遞上來(lái)。

rd,err:=perf.NewReader(events,os.Getpagesize())
iferr!=nil{
log.Fatalf("Creatingperfeventreader:%s",err)
}
deferrd.Close()

//...
vareventpwru.Event
for{
record,err:=rd.Read()
iferr!=nil{
ifperf.IsClosed(err){
return
}
log.Printf("Readingfromperfeventreader:%s",err)
}

ifrecord.LostSamples!=0{
log.Printf("Perfeventringbufferfull,dropped%dsamples",record.LostSamples)
continue
}

iferr:=binary.Read(bytes.NewBuffer(record.RawSample),binary.LittleEndian,&event);err!=nil{
log.Printf("Parsingperfevent:%s",err)
continue
}

output.Print(&event)

select{
case<-ctx.Done():
??break
default:
continue
}
}