-
Notifications
You must be signed in to change notification settings - Fork 24
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Allow to start FLP directly from the flow logs producer
- New "InProcess" ingest stage - New entry point to start the whole FLP in-process: pipeline.StartFLPInProcess - Add some tests
- Loading branch information
Showing
10 changed files
with
339 additions
and
99 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,42 @@ | ||
| package ingest | ||
|
|
||
| import ( | ||
| "github.com/netobserv/flowlogs-pipeline/pkg/config" | ||
| "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils" | ||
| "github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow" | ||
|
|
||
| "github.com/netobserv/netobserv-ebpf-agent/pkg/decode" | ||
| "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| var ilog = logrus.WithField("component", "ingest.InProcess") | ||
|
|
||
| // InProcess ingester, meant to be imported and used from another program via | ||
| type InProcess struct { | ||
| flowPackets chan *pbflow.Records | ||
| } | ||
|
|
||
| func NewInProcess(flowPackets chan *pbflow.Records) *InProcess { | ||
| return &InProcess{flowPackets: flowPackets} | ||
| } | ||
|
|
||
| func (d *InProcess) Ingest(out chan<- config.GenericMap) { | ||
| go func() { | ||
| <-utils.ExitChannel() | ||
| d.Close() | ||
| }() | ||
| for fp := range d.flowPackets { | ||
| ilog.Debugf("Ingested %v records", len(fp.Entries)) | ||
| for _, entry := range fp.Entries { | ||
| out <- decode.PBFlowToMap(entry) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (d *InProcess) Write(record *pbflow.Records) { | ||
| d.flowPackets <- record | ||
| } | ||
|
|
||
| func (d *InProcess) Close() { | ||
| close(d.flowPackets) | ||
| } |
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,32 @@ | ||
| package pipeline | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/netobserv/flowlogs-pipeline/pkg/config" | ||
| "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/ingest" | ||
| "github.com/netobserv/flowlogs-pipeline/pkg/prometheus" | ||
| "github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow" | ||
| ) | ||
|
|
||
| // StartFLPInProcess is an entry point to start the whole FLP / pipeline processing from imported code | ||
| func StartFLPInProcess(cfg config.ConfigFileStruct) (*ingest.InProcess, error) { | ||
| prometheus.SetGlobalMetricsSettings(&cfg.MetricsSettings) | ||
| promServer := prometheus.StartServerAsync(&cfg.MetricsSettings.PromConnectionInfo, nil) | ||
|
|
||
| // Create new flows pipeline | ||
| ingester := ingest.NewInProcess(make(chan *pbflow.Records, 100)) | ||
| flp, err := newPipelineFromIngester(&cfg, ingester) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to initialize pipeline %w", err) | ||
| } | ||
|
|
||
| // Starts the flows pipeline; blocking call | ||
| go func() { | ||
| flp.Run() | ||
| _ = promServer.Shutdown(context.Background()) | ||
| }() | ||
|
|
||
| return ingester, 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,125 @@ | ||
| package pipeline | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "encoding/json" | ||
| "os" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/netobserv/flowlogs-pipeline/pkg/api" | ||
| "github.com/netobserv/flowlogs-pipeline/pkg/config" | ||
| "github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "google.golang.org/protobuf/types/known/durationpb" | ||
| "google.golang.org/protobuf/types/known/timestamppb" | ||
| ) | ||
|
|
||
| func TestInProcessFLP(t *testing.T) { | ||
| pipeline := config.NewPresetIngesterPipeline() | ||
| pipeline = pipeline.WriteStdout("writer", api.WriteStdout{Format: "json"}) | ||
| cfs := config.ConfigFileStruct{ | ||
| Pipeline: pipeline.GetStages(), | ||
| Parameters: pipeline.GetStageParams(), | ||
| } | ||
| ingester, err := StartFLPInProcess(cfs) | ||
| require.NoError(t, err) | ||
| defer ingester.Close() | ||
|
|
||
| capturedOut, w, _ := os.Pipe() | ||
| old := os.Stdout | ||
| os.Stdout = w | ||
| defer func() { | ||
| os.Stdout = old | ||
| }() | ||
|
|
||
| // yield thread to allow pipe services correctly start | ||
| time.Sleep(10 * time.Millisecond) | ||
|
|
||
| startTime := time.Now() | ||
| endTime := startTime.Add(7 * time.Second) | ||
| someDuration := endTime.Sub(startTime) | ||
|
|
||
| ingester.Write(&pbflow.Records{ | ||
| Entries: []*pbflow.Record{{ | ||
| Interface: "eth0", | ||
| EthProtocol: 2048, | ||
| Bytes: 456, | ||
| Packets: 123, | ||
| Direction: pbflow.Direction_EGRESS, | ||
| TimeFlowStart: timestamppb.New(startTime), | ||
| TimeFlowEnd: timestamppb.New(endTime), | ||
| Network: &pbflow.Network{ | ||
| SrcAddr: &pbflow.IP{ | ||
| IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x01020304}, | ||
| }, | ||
| DstAddr: &pbflow.IP{ | ||
| IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x05060708}, | ||
| }, | ||
| Dscp: 1, | ||
| }, | ||
| DataLink: &pbflow.DataLink{ | ||
| DstMac: 0x112233445566, | ||
| SrcMac: 0x010203040506, | ||
| }, | ||
| Transport: &pbflow.Transport{ | ||
| Protocol: 17, | ||
| SrcPort: 23000, | ||
| DstPort: 443, | ||
| }, | ||
| AgentIp: &pbflow.IP{ | ||
| IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a0b0c0d}, | ||
| }, | ||
| PktDropBytes: 100, | ||
| PktDropPackets: 10, | ||
| PktDropLatestFlags: 1, | ||
| PktDropLatestState: 1, | ||
| PktDropLatestDropCause: 8, | ||
| DnsLatency: durationpb.New(someDuration), | ||
| DnsId: 1, | ||
| DnsFlags: 0x80, | ||
| DnsErrno: 0, | ||
| TimeFlowRtt: durationpb.New(someDuration), | ||
| }}, | ||
| }) | ||
|
|
||
| scanner := bufio.NewScanner(capturedOut) | ||
| require.True(t, scanner.Scan()) | ||
| capturedRecord := map[string]interface{}{} | ||
| bytes := scanner.Bytes() | ||
| require.NoError(t, json.Unmarshal(bytes, &capturedRecord), string(bytes)) | ||
|
|
||
| assert.NotZero(t, capturedRecord["TimeReceived"]) | ||
| delete(capturedRecord, "TimeReceived") | ||
| assert.EqualValues(t, map[string]interface{}{ | ||
| "FlowDirection": float64(1), | ||
| "Bytes": float64(456), | ||
| "SrcAddr": "1.2.3.4", | ||
| "DstAddr": "5.6.7.8", | ||
| "Dscp": float64(1), | ||
| "DstMac": "11:22:33:44:55:66", | ||
| "SrcMac": "01:02:03:04:05:06", | ||
| "SrcPort": float64(23000), | ||
| "DstPort": float64(443), | ||
| "Duplicate": false, | ||
| "Etype": float64(2048), | ||
| "Packets": float64(123), | ||
| "Proto": float64(17), | ||
| "TimeFlowStartMs": float64(startTime.UnixMilli()), | ||
| "TimeFlowEndMs": float64(endTime.UnixMilli()), | ||
| "Interface": "eth0", | ||
| "AgentIP": "10.11.12.13", | ||
| "PktDropBytes": float64(100), | ||
| "PktDropPackets": float64(10), | ||
| "PktDropLatestFlags": float64(1), | ||
| "PktDropLatestState": "TCP_ESTABLISHED", | ||
| "PktDropLatestDropCause": "SKB_DROP_REASON_NETFILTER_DROP", | ||
| "DnsLatencyMs": float64(someDuration.Milliseconds()), | ||
| "DnsId": float64(1), | ||
| "DnsFlags": float64(0x80), | ||
| "DnsErrno": float64(0), | ||
| "DnsFlagsResponseCode": "NoError", | ||
| "TimeFlowRttNs": float64(someDuration.Nanoseconds()), | ||
| }, capturedRecord) | ||
| } |
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
Oops, something went wrong.