[FlowAggregator] Use correct input channel in flowExportLoopProxy (#6990

)

The code was invalid as the function was consuming messages straight
from the collector instead of from the preprocessor, hence bypasing the
preprocessor. This was not caught by e2e tests because the preprocessor
output channel is small (capacity of 16), hence it was filling up very
quickly and only a few uninteresting (i.e., irrelevant to the test
cases) records were lost.

Signed-off-by: Antonin Bas <antonin.bas@broadcom.com>

pkg/flowaggregator/flowaggregator.go

@@ -560,7 +560,7 @@ func (fa *flowAggregator) proxyRecord(record ipfixentities.Record, obsDomainID u
 func (fa *flowAggregator) flowExportLoopProxy(stopCh <-chan struct{}) {
 	logTicker := time.NewTicker(fa.logTickerDuration)
 	defer logTicker.Stop()
-	msgCh := fa.collectingProcess.GetMsgChan()
+	msgCh := fa.preprocessorOutCh
 
 	proxyRecords := func(msg *ipfixentities.Message) {
 		set := msg.GetSet()

test/e2e/fixtures.go

@@ -315,8 +315,7 @@ func setupFlowAggregator(tb testing.TB, testData *TestData, o flowVisibilityTest
 		return err
 	}
 	tb.Logf("ClickHouse Service created with ClusterIP: %v", chSvcIP)
-	tb.Logf("Applying flow aggregator YAML with ipfix collector: %s and clickHouse enabled",
-		ipfixCollectorAddr)
+	tb.Logf("Deploying FlowAggregator with ipfix collector: %s and options: %+v", ipfixCollectorAddr, o)
 
 	if err := testData.deployFlowAggregator(ipfixCollectorAddr, o); err != nil {
 		return err