[NETOBSERV-237] gRPC+PB flow ingest/decoder for NetObserv eBPF Agent #160
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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,6 @@ | ||
| package api | ||
|
|
||
| type IngestGRPCProto struct { | ||
| Port int `yaml:"port" doc:"the port number to listen on"` | ||
| BufferLen int `yaml:"buffer_length" doc:"the length of the ingest channel buffer, in groups of flows, containing each group hundreds of flows (default: 100)"` | ||
| } |
This file contains hidden or 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,5 @@ | ||
| package api | ||
|
|
||
| type WriteStdout struct { | ||
| Format string `yaml:"format" doc:"the format of each line: printf (default) or json"` | ||
| } |
This file contains hidden or 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 hidden or 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,85 @@ | ||
| package decode | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "net" | ||
| "time" | ||
|
|
||
| "github.com/netobserv/flowlogs-pipeline/pkg/config" | ||
| "github.com/netobserv/netobserv-agent/pkg/pbflow" | ||
| "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| var pflog = logrus.WithField("component", "Protobuf") | ||
|
|
||
| // Protobuf decodes protobuf flow records definitions, as forwarded by | ||
| // ingest.NetObservAgent, into a Generic Map that follows the same naming conventions | ||
| // as the IPFIX flows from ingest.IngestCollector | ||
| type Protobuf struct { | ||
| } | ||
|
|
||
| func NewProtobuf() (Decoder, error) { | ||
| return &Protobuf{}, nil | ||
| } | ||
|
|
||
| // Decode decodes input strings to a list of flow entries | ||
| func (p *Protobuf) Decode(in []interface{}) []config.GenericMap { | ||
| if len(in) == 0 { | ||
| pflog.Warn("empty input. Skipping") | ||
| return []config.GenericMap{} | ||
| } | ||
| pb, ok := in[0].(*pbflow.Records) | ||
| if !ok { | ||
| pflog.WithField("type", fmt.Sprintf("%T", pb)). | ||
| Warn("expecting input to be *pbflow.Records. Skipping") | ||
| } | ||
| out := make([]config.GenericMap, 0, len(pb.Entries)) | ||
| for _, entry := range pb.Entries { | ||
| out = append(out, pbFlowToMap(entry)) | ||
| } | ||
| return out | ||
| } | ||
|
|
||
| func pbFlowToMap(flow *pbflow.Record) config.GenericMap { | ||
| if flow == nil { | ||
| return config.GenericMap{} | ||
| } | ||
| out := config.GenericMap{ | ||
| "FlowDirection": int(flow.Direction.Number()), | ||
| "Bytes": flow.Bytes, | ||
| "SrcAddr": ipToStr(flow.Network.GetSrcAddr()), | ||
| "DstAddr": ipToStr(flow.Network.GetDstAddr()), | ||
| "SrcMac": macToStr(flow.DataLink.GetSrcMac()), | ||
| "DstMac": macToStr(flow.DataLink.GetDstMac()), | ||
| "SrcPort": flow.Transport.GetSrcPort(), | ||
| "DstPort": flow.Transport.GetDstPort(), | ||
| "Etype": flow.EthProtocol, | ||
| "Packets": flow.Packets, | ||
| "Proto": flow.Transport.GetProtocol(), | ||
| "TimeFlowStart": flow.TimeFlowStart.GetSeconds(), | ||
| "TimeFlowEnd": flow.TimeFlowEnd.GetSeconds(), | ||
| "TimeReceived": time.Now().Unix(), | ||
| "Interface": flow.Interface, | ||
| } | ||
| return out | ||
| } | ||
|
|
||
| func ipToStr(ip *pbflow.IP) string { | ||
|
Collaborator
There was a problem hiding this comment. If
Author
There was a problem hiding this comment. I kept them as private functions because they rely on very concrete implementations of the encoding, as they were defined by us. I can move it to public functions if you think it's better to do it, but maybe they won't be used anywhere else and will increase the surface of the public API.
Collaborator
There was a problem hiding this comment. @mariomac ok ... so if they are specific to the protocol you are correct and they should be kept in this context ... the only sub-idea would be to move to a file called |
||
| if ip.GetIpv6() != nil { | ||
| return net.IP(ip.GetIpv6()).String() | ||
| } else { | ||
| n := ip.GetIpv4() | ||
| return fmt.Sprintf("%d.%d.%d.%d", | ||
| byte(n>>24), byte(n>>16), byte(n>>8), byte(n)) | ||
| } | ||
| } | ||
|
|
||
| func macToStr(mac uint64) string { | ||
| return fmt.Sprintf("%02X:%02X:%02X:%02X:%02X:%02X", | ||
| uint8(mac>>40), | ||
| uint8(mac>>32), | ||
| uint8(mac>>24), | ||
| uint8(mac>>16), | ||
| uint8(mac>>8), | ||
| uint8(mac)) | ||
| } | ||
This file contains hidden or 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,66 @@ | ||
| package decode | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/netobserv/flowlogs-pipeline/pkg/config" | ||
| "github.com/netobserv/netobserv-agent/pkg/pbflow" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "google.golang.org/protobuf/types/known/timestamppb" | ||
| ) | ||
|
|
||
| func TestDecodePBFlows(t *testing.T) { | ||
| decoder := Protobuf{} | ||
|
|
||
| someTime := time.Now() | ||
| flow := &pbflow.Record{ | ||
| Interface: "eth0", | ||
| EthProtocol: 2048, | ||
| Bytes: 456, | ||
| Packets: 123, | ||
| Direction: pbflow.Direction_EGRESS, | ||
| TimeFlowStart: timestamppb.New(someTime), | ||
| TimeFlowEnd: timestamppb.New(someTime), | ||
| Network: &pbflow.Network{ | ||
| SrcAddr: &pbflow.IP{ | ||
| IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x01020304}, | ||
| }, | ||
| DstAddr: &pbflow.IP{ | ||
| IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x05060708}, | ||
| }, | ||
| }, | ||
| DataLink: &pbflow.DataLink{ | ||
| DstMac: 0x112233445566, | ||
| SrcMac: 0x010203040506, | ||
| }, | ||
| Transport: &pbflow.Transport{ | ||
| Protocol: 1, | ||
| SrcPort: 23000, | ||
| DstPort: 443, | ||
| }, | ||
| } | ||
|
|
||
| out := decoder.Decode([]interface{}{&pbflow.Records{Entries: []*pbflow.Record{flow}}}) | ||
| require.Len(t, out, 1) | ||
| assert.NotZero(t, out[0]["TimeReceived"]) | ||
| delete(out[0], "TimeReceived") | ||
| assert.Equal(t, config.GenericMap{ | ||
| "FlowDirection": 1, | ||
| "Bytes": uint64(456), | ||
| "SrcAddr": "1.2.3.4", | ||
| "DstAddr": "5.6.7.8", | ||
| "DstMac": "11:22:33:44:55:66", | ||
| "SrcMac": "01:02:03:04:05:06", | ||
| "SrcPort": uint32(23000), | ||
| "DstPort": uint32(443), | ||
| "Etype": uint32(2048), | ||
| "Packets": uint64(123), | ||
| "Proto": uint32(1), | ||
| "TimeFlowStart": someTime.Unix(), | ||
| "TimeFlowEnd": someTime.Unix(), | ||
| "Interface": "eth0", | ||
| }, out[0]) | ||
|
|
||
| } |
This file contains hidden or 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,49 @@ | ||
| package ingest | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/netobserv/flowlogs-pipeline/pkg/config" | ||
| "github.com/netobserv/netobserv-agent/pkg/grpc" | ||
| "github.com/netobserv/netobserv-agent/pkg/pbflow" | ||
| ) | ||
|
|
||
| const defaultBufferLen = 100 | ||
|
|
||
| // GRPCProtobuf ingests data from the NetObserv eBPF Agent, using Protocol Buffers over gRPC | ||
| type GRPCProtobuf struct { | ||
| collector *grpc.CollectorServer | ||
| flowPackets chan *pbflow.Records | ||
| } | ||
|
|
||
| func NewGRPCProtobuf(params config.StageParam) (*GRPCProtobuf, error) { | ||
| netObserv := params.Ingest.GRPC | ||
| if netObserv.Port == 0 { | ||
| return nil, fmt.Errorf("ingest port not specified") | ||
| } | ||
| bufLen := netObserv.BufferLen | ||
| if bufLen == 0 { | ||
| bufLen = defaultBufferLen | ||
| } | ||
| flowPackets := make(chan *pbflow.Records, bufLen) | ||
| collector, err := grpc.StartCollector(netObserv.Port, flowPackets) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &GRPCProtobuf{ | ||
| collector: collector, | ||
| flowPackets: flowPackets, | ||
| }, nil | ||
| } | ||
|
|
||
| func (no *GRPCProtobuf) Ingest(out chan<- []interface{}) { | ||
| for fp := range no.flowPackets { | ||
| out <- []interface{}{fp} | ||
| } | ||
| } | ||
|
|
||
| func (no *GRPCProtobuf) Close() error { | ||
| err := no.collector.Close() | ||
| close(no.flowPackets) | ||
| return err | ||
| } |
This file contains hidden or 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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should this be built somehow dynamic from https://github.com/netobserv/netobserv-agent/blob/main/proto/flow.proto ????
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That would involve flattening/changing the
flow.protodefinition to match the map structure, and then create some sort of code generation or use reflection to create the map. Since the number of fields is still relatively small, I'd keep using this explicit conversion, and consider another approach in the future if we feel that the fields grow until making this conversion unmaintainable.Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@mariomac this is exactly my fear :-( .... that this will grow and no one will be able to find where this came from and why this is not working. Any chance to someone connect that to the source (i.e. to the https://github.com/netobserv/netobserv-agent) project ... if not in code then at least with some remarks maybe?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah, actually the
*pbflow.Recordstype is imported from thegithub.com/netobserv/netobserv-agentdirectly, so IDE navigation leads you to the original protobuf definition (at least in Goland IDE).But I agree that we should investigate a more automatic way to map protobuf --> generic map fields that does not penalize readability nor performance.