-
Notifications
You must be signed in to change notification settings - Fork 108
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Network metrics: filter by transport protocol (#752)
* move transport.Protocol to its own package * protocol filter node * Integrate protocol filter into pipeline * optimized and cleaned up allower/excluder interfacing
- Loading branch information
Showing
10 changed files
with
381 additions
and
77 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| package flow | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/mariomac/pipes/pipe" | ||
|
|
||
| "github.com/grafana/beyla/pkg/internal/netolly/ebpf" | ||
| "github.com/grafana/beyla/pkg/internal/netolly/flow/transport" | ||
| ) | ||
|
|
||
| // ProtocolFilterProvider allows selecting which protocols are going to be instrumented. | ||
| // It drops any flow not appearing in the "allowed" list. | ||
| // If the Allowed list is empty, it drops any flow appearing in the "excluded" list. | ||
| func ProtocolFilterProvider(allowed, excluded []string) pipe.MiddleProvider[[]*ebpf.Record, []*ebpf.Record] { | ||
| return func() (pipe.MiddleFunc[[]*ebpf.Record, []*ebpf.Record], error) { | ||
| if len(allowed) == 0 && len(excluded) == 0 { | ||
| // user did not configure any filter. Ignore this node | ||
| return pipe.Bypass[[]*ebpf.Record](), nil | ||
| } | ||
| pf, err := newFilter(allowed, excluded) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return pf.nodeLoop, nil | ||
| } | ||
| } | ||
|
|
||
| type protocolFilter struct { | ||
| isAllowed func(r *ebpf.Record) bool | ||
| } | ||
|
|
||
| func newFilter(allowed, excluded []string) (*protocolFilter, error) { | ||
| // if the allowed list has items, only interfaces in that list are allowed | ||
| if len(allowed) > 0 { | ||
| allow, err := allower(allowed) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &protocolFilter{isAllowed: allow}, nil | ||
| } | ||
| // if the allowed list is empty, any interface is allowed except if it matches the exclusion list | ||
| exclude, err := excluder(excluded) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &protocolFilter{isAllowed: exclude}, nil | ||
| } | ||
|
|
||
| func (pf *protocolFilter) nodeLoop(in <-chan []*ebpf.Record, out chan<- []*ebpf.Record) { | ||
| for records := range in { | ||
| if filtered := pf.filter(records); len(filtered) > 0 { | ||
| out <- filtered | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (pf *protocolFilter) filter(input []*ebpf.Record) []*ebpf.Record { | ||
| writeIdx := 0 | ||
| for readIdx := range input { | ||
| if pf.isAllowed(input[readIdx]) { | ||
| input[writeIdx] = input[readIdx] | ||
| writeIdx++ | ||
| } | ||
| } | ||
| return input[:writeIdx] | ||
| } | ||
|
|
||
| func allower(allowed []string) (func(r *ebpf.Record) bool, error) { | ||
| allow, err := protocolsMap(allowed) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("in network protocols: %w", err) | ||
| } | ||
| return func(r *ebpf.Record) bool { | ||
| _, ok := allow[transport.Protocol(r.Id.TransportProtocol)] | ||
| return ok | ||
| }, nil | ||
| } | ||
|
|
||
| func excluder(excluded []string) (func(r *ebpf.Record) bool, error) { | ||
| exclude, err := protocolsMap(excluded) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("in network excluded protocols: %w", err) | ||
| } | ||
| return func(r *ebpf.Record) bool { | ||
| _, excluded := exclude[transport.Protocol(r.Id.TransportProtocol)] | ||
| return !excluded | ||
| }, nil | ||
| } | ||
|
|
||
| func protocolsMap(entries []string) (map[transport.Protocol]struct{}, error) { | ||
| protoMap := map[transport.Protocol]struct{}{} | ||
| for _, aStr := range entries { | ||
| if atp, err := transport.ParseProtocol(aStr); err == nil { | ||
| protoMap[atp] = struct{}{} | ||
| } else { | ||
| return nil, err | ||
| } | ||
| } | ||
| return protoMap, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| package flow | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/grafana/beyla/pkg/internal/netolly/ebpf" | ||
| "github.com/grafana/beyla/pkg/internal/netolly/flow/transport" | ||
| "github.com/grafana/beyla/pkg/internal/testutil" | ||
| ) | ||
|
|
||
| var tcp1 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 1, TransportProtocol: uint8(transport.TCP)}}} | ||
| var tcp2 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 2, TransportProtocol: uint8(transport.TCP)}}} | ||
| var tcp3 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 3, TransportProtocol: uint8(transport.TCP)}}} | ||
| var udp1 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 4, TransportProtocol: uint8(transport.UDP)}}} | ||
| var udp2 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 5, TransportProtocol: uint8(transport.UDP)}}} | ||
| var icmp1 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 7, TransportProtocol: uint8(transport.ICMP)}}} | ||
| var icmp2 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 8, TransportProtocol: uint8(transport.ICMP)}}} | ||
| var icmp3 = &ebpf.Record{NetFlowRecordT: ebpf.NetFlowRecordT{Id: ebpf.NetFlowId{SrcPort: 9, TransportProtocol: uint8(transport.ICMP)}}} | ||
|
|
||
| func TestProtocolFilter_Allow(t *testing.T) { | ||
| protocolFilter, err := ProtocolFilterProvider([]string{"TCP"}, nil)() | ||
| require.NoError(t, err) | ||
| input, output := make(chan []*ebpf.Record, 10), make(chan []*ebpf.Record, 10) | ||
| defer close(input) | ||
| go protocolFilter(input, output) | ||
|
|
||
| input <- []*ebpf.Record{} | ||
| input <- []*ebpf.Record{tcp1, tcp2, tcp3} | ||
| input <- []*ebpf.Record{icmp2, udp1, icmp1, udp2, icmp3} | ||
| input <- []*ebpf.Record{icmp2, tcp1, udp1, icmp1, tcp2, udp2, tcp3, icmp3} | ||
|
|
||
| filtered := testutil.ReadChannel(t, output, timeout) | ||
| assert.Equal(t, []*ebpf.Record{tcp1, tcp2, tcp3}, filtered) | ||
| filtered = testutil.ReadChannel(t, output, timeout) | ||
| assert.Equal(t, []*ebpf.Record{tcp1, tcp2, tcp3}, filtered) | ||
| // no more slices are sent (the second was completely filtered) | ||
| select { | ||
| case o := <-output: | ||
| require.Failf(t, "unexpected flows!", "%v", o) | ||
| default: | ||
| // ok!! | ||
| } | ||
| } | ||
|
|
||
| func TestProtocolFilter_Exclude(t *testing.T) { | ||
| protocolFilter, err := ProtocolFilterProvider(nil, []string{"TCP"})() | ||
| require.NoError(t, err) | ||
| input, output := make(chan []*ebpf.Record, 10), make(chan []*ebpf.Record, 10) | ||
| defer close(input) | ||
| go protocolFilter(input, output) | ||
|
|
||
| input <- []*ebpf.Record{tcp1, tcp2, tcp3} | ||
| input <- []*ebpf.Record{icmp2, udp1, icmp1, udp2, icmp3} | ||
| input <- []*ebpf.Record{} | ||
| input <- []*ebpf.Record{icmp2, tcp1, udp1, icmp1, tcp2, udp2, tcp3, icmp3} | ||
|
|
||
| filtered := testutil.ReadChannel(t, output, timeout) | ||
| assert.Equal(t, []*ebpf.Record{icmp2, udp1, icmp1, udp2, icmp3}, filtered) | ||
| filtered = testutil.ReadChannel(t, output, timeout) | ||
| assert.Equal(t, []*ebpf.Record{icmp2, udp1, icmp1, udp2, icmp3}, filtered) | ||
| // no more slices are sent (the first was completely filtered) | ||
| select { | ||
| case o := <-output: | ||
| require.Failf(t, "unexpected flows!", "%v", o) | ||
| default: | ||
| // ok!! | ||
| } | ||
| } | ||
| func TestProtocolFilter_ParsingErrors(t *testing.T) { | ||
| _, err := ProtocolFilterProvider([]string{"TCP", "tralara"}, nil)() | ||
| assert.Error(t, err) | ||
| _, err = ProtocolFilterProvider([]string{"TCP", "tralara"}, []string{"UDP"})() | ||
| assert.Error(t, err) | ||
| _, err = ProtocolFilterProvider(nil, []string{"TCP", "tralara"})() | ||
| assert.Error(t, err) | ||
| } |
Oops, something went wrong.