Bug fix for inter-node traffic

In this commit, we do:

1. Set ReadyToSend to true for flows don’t need correlation.
2. Set areCorrelatedFieldsFilled to true for flows don’t belong to inter-node traffic. For flows need to do correlation, its areCorrelatedFieldsFilled will be set to true once the correlation job is finished.

Signed-off-by: Yun-Tang Hsu <[email protected]>

pkg/intermediate/aggregate.go

@@ -343,6 +343,20 @@ func (a *AggregationProcess) addOrUpdateRecordInMap(flowKey *FlowKey, record ent
 
 	currTime := time.Now()
 	aggregationRecord, exist := a.flowKeyRecordMap[*flowKey]
+	if exist {
+		if flowTypeIE, _, exist := aggregationRecord.Record.GetInfoElementWithValue("flowType"); exist {
+			flowType = flowTypeIE.GetUnsigned8Value()
+		} else {
+			klog.Warning("FlowType does not exist in current record.")
+		}
+		prevCorrelationRequired := isCorrelationRequired(flowType, aggregationRecord.Record)
+		if prevCorrelationRequired && !correlationRequired {
+			delete(a.flowKeyRecordMap, *flowKey)
+			exist = false
+		} else if !prevCorrelationRequired && correlationRequired {
+			return nil
+		}
+	}
 	if exist {
 		if correlationRequired {
 			// Do correlation of records if record belongs to inter-node flow and
@@ -408,17 +422,10 @@ func (a *AggregationProcess) addOrUpdateRecordInMap(flowKey *FlowKey, record ent
 			waitForReadyToSendRetries: 0,
 			isIPv4:                    isIPv4,
 		}
-
-		if !correlationRequired {
-			aggregationRecord.ReadyToSend = true
-			// If no correlation is required for an Inter-Node record, K8s metadata is
-			// expected to be not completely filled. For Intra-Node flows and ToExternal
-			// flows, areCorrelatedFieldsFilled is set to true by default.
-			if flowType == registry.FlowTypeInterNode {
-				aggregationRecord.areCorrelatedFieldsFilled = false
-			} else {
-				aggregationRecord.areCorrelatedFieldsFilled = true
-			}
+		// For intra-node and external traffic, areCorrelatedFieldsFilled is always true.
+		// For inter-node with allow np/anp action, its areExternalFieldsFilled will be set to true once the correlation job is finished.
+		if flowType != registry.FlowTypeInterNode {
+			aggregationRecord.areCorrelatedFieldsFilled = true
 		}
 		aggregationRecord.areExternalFieldsFilled = false
 		// Push the record to the priority queue.
@@ -432,6 +439,10 @@ func (a *AggregationProcess) addOrUpdateRecordInMap(flowKey *FlowKey, record ent
 		pqItem.inactiveExpireTime = currTime.Add(a.inactiveExpiryTimeout)
 		heap.Push(&a.expirePriorityQueue, pqItem)
 	}
+	// For flows that do not need correlation, ReadyToSend should always be true.
+	if !correlationRequired {
+		aggregationRecord.ReadyToSend = true
+	}
 	a.flowKeyRecordMap[*flowKey] = aggregationRecord
 	return nil
 }