github/portmaster
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Improve epbf bandwidth stats monitor

Browse Source
This commit is contained in:
Daniel
2023-07-20 14:02:21 +02:00
parent 4c21c87b8a
commit 6d569ca346
6 changed files with 121 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
firewall/interception/ebpf/bandwidth/bpf_bpfeb.go
View File

@@ -14,8 +14,9 @@ import (
)
type bpfSkInfo struct {
Rx uint64
Tx uint64
Rx uint64
Tx uint64
Reported uint64
}
type bpfSkKey struct {

BIN
firewall/interception/ebpf/bandwidth/bpf_bpfeb.o
View File

Binary file not shown.

5
firewall/interception/ebpf/bandwidth/bpf_bpfel.go
View File

@@ -14,8 +14,9 @@ import (
)
type bpfSkInfo struct {
Rx uint64
Tx uint64
Rx uint64
Tx uint64
Reported uint64
}
type bpfSkKey struct {

BIN
firewall/interception/ebpf/bandwidth/bpf_bpfel.o
View File

Binary file not shown.

194
firewall/interception/ebpf/bandwidth/interface.go
View File

@@ -1,148 +1,152 @@
package ebpf
import (
"context"
"encoding/binary"
"fmt"
"net"
"path/filepath"
"sync/atomic"
"syscall"
"time"
"github.com/cilium/ebpf"
"github.com/cilium/ebpf/link"
"github.com/cilium/ebpf/rlimit"
"golang.org/x/sys/unix"
"github.com/safing/portbase/log"
"github.com/safing/portmaster/network/packet"
"golang.org/x/sys/unix"
)
//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cc clang -cflags "-O2 -g -Wall -Werror" bpf ../programs/bandwidth.c
var ebpfInterface = struct {
objs bpfObjects
sockOptionsLink link.Link
udpv4SMLink link.Link
udpv4RMLink link.Link
udpv6SMLink link.Link
udpv6RMLink link.Link
}{
objs: bpfObjects{},
}
// SetupBandwidthInterface initializes the ebpf interface and starts gattering bandwidth information for all connections.
func SetupBandwidthInterface() error {
var ebpfLoadingFailed atomic.Uint32
// BandwidthStatsWorker monitors connection bandwidth using ebpf.
func BandwidthStatsWorker(ctx context.Context, collectInterval time.Duration, bandwidthUpdates chan *packet.BandwidthUpdate) error {
// Allow the current process to lock memory for eBPF resources.
err := rlimit.RemoveMemlock()
if err != nil {
return fmt.Errorf("failed to remove memlock: %s", err)
return fmt.Errorf("ebpf: failed to remove memlock: %w", err)
}
// Load pre-compiled programs and maps into the kernel.
err = loadBpfObjects(&ebpfInterface.objs, nil)
if err != nil {
return fmt.Errorf("feiled loading objects: %s", err)
}
defer func() {
if err != nil {
// Defer the cleanup function to be called at the end of the enclosing function
// If there was an error during the execution, shutdown the BandwithInterface
ShutdownBandwithInterface()
objs := bpfObjects{}
if err := loadBpfObjects(&objs, nil); err != nil {
if ebpfLoadingFailed.Add(1) >= 5 {
log.Warningf("ebpf: failed to load ebpf object 5 times, giving up with error %s", err)
return nil
}
}()
return fmt.Errorf("ebpf: failed to load ebpf object: %w", err)
}
defer objs.Close() //nolint:errcheck
// Find the cgroup path
path, err := findCgroupPath()
if err != nil {
return fmt.Errorf("faield to find cgroup paths: %s", err)
return fmt.Errorf("ebpf: failed to find cgroup paths: %w", err)
}
// Attach socket options for monitoring connections
ebpfInterface.sockOptionsLink, err = link.AttachCgroup(link.CgroupOptions{
sockOptionsLink, err := link.AttachCgroup(link.CgroupOptions{
Path: path,
Program: ebpfInterface.objs.bpfPrograms.SocketOperations,
Program: objs.bpfPrograms.SocketOperations,
Attach: ebpf.AttachCGroupSockOps,
})
if err != nil {
return fmt.Errorf("Failed to open module sockops: %s", err)
return fmt.Errorf("ebpf: failed to open module sockops: %w", err)
}
defer sockOptionsLink.Close() //nolint:errcheck
// Attach Udp Ipv4 recive message tracing
ebpfInterface.udpv4RMLink, err = link.AttachTracing(link.TracingOptions{
Program: ebpfInterface.objs.UdpRecvmsg,
udpv4RMLink, err := link.AttachTracing(link.TracingOptions{
Program: objs.bpfPrograms.UdpRecvmsg,
})
if err != nil {
return fmt.Errorf("Failed to open trace Udp IPv4 recvmsg: %s", err)
return fmt.Errorf("ebpf: failed to open trace Udp IPv4 recvmsg: %w", err)
}
defer udpv4RMLink.Close() //nolint:errcheck
// Attach UDP IPv4 send message tracing
ebpfInterface.udpv4SMLink, err = link.AttachTracing(link.TracingOptions{
Program: ebpfInterface.objs.UdpSendmsg,
udpv4SMLink, err := link.AttachTracing(link.TracingOptions{
Program: objs.bpfPrograms.UdpSendmsg,
})
if err != nil {
return fmt.Errorf("Failed to open trace Udp IPv4 sendmsg: %s", err)
return fmt.Errorf("ebpf: failed to open trace Udp IPv4 sendmsg: %w", err)
}
defer udpv4SMLink.Close() //nolint:errcheck
// Attach UDP IPv6 receive message tracing
ebpfInterface.udpv6RMLink, err = link.AttachTracing(link.TracingOptions{
Program: ebpfInterface.objs.Udpv6Recvmsg,
udpv6RMLink, err := link.AttachTracing(link.TracingOptions{
Program: objs.bpfPrograms.Udpv6Recvmsg,
})
if err != nil {
return fmt.Errorf("Failed to open trace Udp IPv6 recvmsg: %s", err)
return fmt.Errorf("ebpf: failed to open trace Udp IPv6 recvmsg: %w", err)
}
defer udpv6RMLink.Close() //nolint:errcheck
// Attach UDP IPv6 send message tracing
ebpfInterface.udpv6RMLink, err = link.AttachTracing(link.TracingOptions{
Program: ebpfInterface.objs.Udpv6Sendmsg,
udpv6SMLink, err := link.AttachTracing(link.TracingOptions{
Program: objs.bpfPrograms.Udpv6Sendmsg,
})
if err != nil {
return fmt.Errorf("Failed to open trace Udp IPv6 sendmsg: %s", err)
return fmt.Errorf("ebpf: failed to open trace Udp IPv6 sendmsg: %w", err)
}
defer udpv6SMLink.Close() //nolint:errcheck
// Example code that will print the bandwidth table every 10 seconds
// go func() {
// ticker := time.NewTicker(10 * time.Second)
// defer ticker.Stop()
// for range ticker.C {
// printBandwidthData()
// }
// }()
// Setup ticker.
ticker := time.NewTicker(collectInterval)
defer ticker.Stop()
return nil
// Collect bandwidth at every tick.
for {
select {
case <-ticker.C:
reportBandwidth(ctx, objs, bandwidthUpdates)
case <-ctx.Done():
return nil
}
}
}
// ShutdownBandwithInterface shuts down the bandwidth interface by closing the associated links and objects.
func ShutdownBandwithInterface() {
// Close the sockOptionsLink if it is not nil
if ebpfInterface.sockOptionsLink != nil {
ebpfInterface.sockOptionsLink.Close()
}
// reportBandwidth reports the bandwidth to the given updates channel.
func reportBandwidth(ctx context.Context, objs bpfObjects, bandwidthUpdates chan *packet.BandwidthUpdate) {
iter := objs.bpfMaps.PmBandwidthMap.Iterate()
var skKey bpfSkKey
var skInfo bpfSkInfo
for iter.Next(&skKey, &skInfo) {
// Check if already reported.
if skInfo.Reported >= 1 {
continue
}
// Mark as reported and update the map.
skInfo.Reported = 1
if err := objs.bpfMaps.PmBandwidthMap.Put(&skKey, &skInfo); err != nil {
log.Debugf("ebpf: failed to update map: %s", err)
}
// Close the udpv4SMLink if it is not nil
if ebpfInterface.udpv4SMLink != nil {
ebpfInterface.udpv4SMLink.Close()
connID := packet.CreateConnectionID(
packet.IPProtocol(skKey.Protocol),
convertArrayToIP(skKey.SrcIp, skKey.Ipv6 == 1), skKey.SrcPort,
convertArrayToIP(skKey.DstIp, skKey.Ipv6 == 1), skKey.DstPort,
false,
)
update := &packet.BandwidthUpdate{
ConnID: connID,
RecvBytes: skInfo.Rx,
SentBytes: skInfo.Tx,
Method: packet.Absolute,
}
select {
case bandwidthUpdates <- update:
case <-ctx.Done():
return
}
}
// Close the udpv4RMLink if it is not nil
if ebpfInterface.udpv4RMLink != nil {
ebpfInterface.udpv4RMLink.Close()
}
// Close the udpv6SMLink if it is not nil
if ebpfInterface.udpv6SMLink != nil {
ebpfInterface.udpv6SMLink.Close()
}
// Close the udpv6RMLink if it is not nil
if ebpfInterface.udpv6RMLink != nil {
ebpfInterface.udpv6RMLink.Close()
}
// Close the ebpfInterface objects
ebpfInterface.objs.Close()
}
// findCgroupPath returns the default unified path of the cgroup
// findCgroupPath returns the default unified path of the cgroup.
func findCgroupPath() (string, error) {
cgroupPath := "/sys/fs/cgroup"
@@ -158,31 +162,17 @@ func findCgroupPath() (string, error) {
return cgroupPath, nil
}
// printBandwidthData prints the contencs of the shared map in the ebpf program.
func printBandwidthData() {
iter := ebpfInterface.objs.bpfMaps.PmBandwidthMap.Iterate()
var skKey bpfSkKey
var skInfo bpfSkInfo
for iter.Next(&skKey, &skInfo) {
log.Debugf("Connection: %d %s:%d %s:%d %d %d", skKey.Protocol,
convertArrayToIPv4(skKey.SrcIp, packet.IPVersion(skKey.Ipv6)).String(), skKey.SrcPort,
convertArrayToIPv4(skKey.DstIp, packet.IPVersion(skKey.Ipv6)).String(), skKey.DstPort,
skInfo.Rx, skInfo.Tx,
)
}
}
// convertArrayToIPv4 converts an array of uint32 values to an IPv4 net.IP address.
func convertArrayToIPv4(input [4]uint32, ipVersion packet.IPVersion) net.IP {
if ipVersion == packet.IPv4 {
// convertArrayToIP converts an array of uint32 values to a net.IP address.
func convertArrayToIP(input [4]uint32, ipv6 bool) net.IP {
if !ipv6 {
addressBuf := make([]byte, 4)
binary.LittleEndian.PutUint32(addressBuf, input[0])
return net.IP(addressBuf)
} else {
addressBuf := make([]byte, 16)
for i := 0; i < 4; i++ {
binary.LittleEndian.PutUint32(addressBuf[i*4:i*4+4], input[i])
}
return net.IP(addressBuf)
}
addressBuf := make([]byte, 16)
for i := 0; i < 4; i++ {
binary.LittleEndian.PutUint32(addressBuf[i*4:i*4+4], input[i])
}
return net.IP(addressBuf)
}

32
firewall/interception/ebpf/programs/bandwidth.c
View File

@@ -9,7 +9,7 @@
#define PROTOCOL_TCP 6
#define PROTOCOL_UDP 17
char __license[] SEC("license") = "Dual MIT/GPL";
char __license[] SEC("license") = "GPL";
struct sk_key {
u32 src_ip[4];
@@ -23,6 +23,7 @@ struct sk_key {
struct sk_info {
u64 rx;
u64 tx;
u64 reported;
};
// Max number of connections that will be kept. Increse the number if it's not enough.
@@ -84,12 +85,21 @@ int socket_operations(struct bpf_sock_ops *skops) {
}
key.dst_port = __builtin_bswap16(sk->dst_port);
key.ipv6 = 1;
// FIXME: This should be added here too, but loading the ebpf module fails if we add it.
// This 100% the same thing as above. No clue, man.
// struct sk_info newInfo = {0};
// newInfo.rx = skops->bytes_received;
// newInfo.tx = skops->bytes_acked;
// bpf_map_update_elem(&pm_bandwidth_map, &key, &newInfo, BPF_ANY);
}
return 0;
}
// udp_sendmsg hookes to the equvelent kernel function and saves the bandwoth data
// udp_sendmsg hookes to the respective kernel function and saves the bandwidth data
SEC("fentry/udp_sendmsg")
int BPF_PROG(udp_sendmsg, struct sock *sk, struct msghdr *msg, size_t len) {
struct sock_common *skc = &sk->__sk_common;
@@ -106,7 +116,8 @@ int BPF_PROG(udp_sendmsg, struct sock *sk, struct msghdr *msg, size_t len) {
// Update the map with the new information
struct sk_info *info = bpf_map_lookup_elem(&pm_bandwidth_map, &key);
if (info != NULL) {
__sync_fetch_and_add(&info->tx, len);
__sync_fetch_and_add(&info->tx, len); // TODO: Use atomic instead.
__sync_fetch_and_and(&info->reported, 0); // TODO: Use atomic instead.
} else {
struct sk_info newInfo = {0};
@@ -117,7 +128,7 @@ int BPF_PROG(udp_sendmsg, struct sock *sk, struct msghdr *msg, size_t len) {
return 0;
};
// udp_recvmsg hookes to the equvelent kernel function and saves the bandwoth data
// udp_recvmsg hookes to the respective kernel function and saves the bandwidth data
SEC("fentry/udp_recvmsg")
int BPF_PROG(udp_recvmsg, struct sock *sk, struct msghdr *msg, size_t len, int flags, int *addr_len) {
struct sock_common *skc = &sk->__sk_common;
@@ -134,7 +145,8 @@ int BPF_PROG(udp_recvmsg, struct sock *sk, struct msghdr *msg, size_t len, int f
// Update the map with the new information
struct sk_info *info = bpf_map_lookup_elem(&pm_bandwidth_map, &key);
if (info != NULL) {
__sync_fetch_and_add(&info->rx, len);
__sync_fetch_and_add(&info->rx, len); // TODO: Use atomic instead.
__sync_fetch_and_and(&info->reported, 0); // TODO: Use atomic instead.
} else {
struct sk_info newInfo = {0};
@@ -145,7 +157,7 @@ int BPF_PROG(udp_recvmsg, struct sock *sk, struct msghdr *msg, size_t len, int f
return 0;
};
// udpv6_sendmsg hookes to the equvelent kernel function and saves the bandwoth data
// udpv6_sendmsg hookes to the respective kernel function and saves the bandwidth data
SEC("fentry/udpv6_sendmsg")
int BPF_PROG(udpv6_sendmsg, struct sock *sk, struct msghdr *msg, size_t len) {
struct sock_common *skc = &sk->__sk_common;
@@ -164,7 +176,8 @@ int BPF_PROG(udpv6_sendmsg, struct sock *sk, struct msghdr *msg, size_t len) {
// Update the map with the new information
struct sk_info *info = bpf_map_lookup_elem(&pm_bandwidth_map, &key);
if (info != NULL) {
__sync_fetch_and_add(&info->tx, len);
__sync_fetch_and_add(&info->tx, len); // TODO: Use atomic instead.
__sync_fetch_and_and(&info->reported, 0); // TODO: Use atomic instead.
} else {
struct sk_info newInfo = {0};
newInfo.tx = len;
@@ -174,7 +187,7 @@ int BPF_PROG(udpv6_sendmsg, struct sock *sk, struct msghdr *msg, size_t len) {
return 0;
}
// udpv6_recvmsg hookes to the equvelent kernel function and saves the bandwoth data
// udpv6_recvmsg hookes to the respective kernel function and saves the bandwidth data
SEC("fentry/udpv6_recvmsg")
int BPF_PROG(udpv6_recvmsg, struct sock *sk, struct msghdr *msg, size_t len, int flags, int *addr_len) {
struct sock_common *skc = &sk->__sk_common;
@@ -193,7 +206,8 @@ int BPF_PROG(udpv6_recvmsg, struct sock *sk, struct msghdr *msg, size_t len, int
// Update the map with the new information
struct sk_info *info = bpf_map_lookup_elem(&pm_bandwidth_map, &key);
if (info != NULL) {
__sync_fetch_and_add(&info->rx, len);
__sync_fetch_and_add(&info->rx, len); // TODO: Use atomic instead.
__sync_fetch_and_and(&info->reported, 0); // TODO: Use atomic instead.
} else {
struct sk_info newInfo = {0};
newInfo.rx = len;