Fix the deadlock between exporter and conntrack polling go routines

Deadlock is due to the access of connection map to update from exporter
go routine to update the "DoneExport" flag. This was caught in scale testing.
Resolved this through a temporary fix by adding same flag in record
data struct. The connection and record deletion logic will be re-evaluated
through PR antrea-io#2360 as it refactors the related code.

pkg/agent/flowexporter/connections/conntrack_connections.go

@@ -247,16 +247,3 @@ func (cs *ConntrackConnectionStore) DeleteConnWithoutLock(connKey flowexporter.C
 	metrics.TotalAntreaConnectionsInConnTrackTable.Dec()
 	return nil
 }
-
-// SetExportDone sets DoneExport field of conntrack connection to true given the connection key.
-func (cs *ConntrackConnectionStore) SetExportDone(connKey flowexporter.ConnectionKey) error {
-	cs.mutex.Lock()
-	defer cs.mutex.Unlock()
-
-	if conn, found := cs.connections[connKey]; !found {
-		return fmt.Errorf("connection with key %v does not exist in connection map", connKey)
-	} else {
-		conn.DoneExport = true
-		return nil
-	}
-}

pkg/agent/flowexporter/exporter/exporter.go

@@ -302,14 +302,9 @@ func (exp *flowExporter) sendFlowRecords() error {
 
 			if flowexporter.IsConnectionDying(&record.Conn) {
 				// If the connection is in dying state or connection is not in conntrack table,
-				// we will delete the flow records from records map.
+				// we set the DoneExport flag to do the deletion later.
 				klog.V(2).Infof("Deleting the inactive flow records with key: %v from record map", key)
-				if err := exp.flowRecords.DeleteFlowRecordWithoutLock(key); err != nil {
-					return err
-				}
-				if err := exp.conntrackConnStore.SetExportDone(key); err != nil {
-					return err
-				}
+				record.DoneExport = true
 			} else {
 				exp.flowRecords.ValidateAndUpdateStats(key, record)
 			}

pkg/agent/flowexporter/flowrecords/flow_records.go

@@ -39,8 +39,8 @@ func NewFlowRecords() *FlowRecords {
 // It makes a copy of the connection object to record, to avoid race conditions between the
 // connection store and the flow exporter.
 func (fr *FlowRecords) AddOrUpdateFlowRecord(key flowexporter.ConnectionKey, conn *flowexporter.Connection) error {
-	// If the connection is in dying state and the corresponding flow records are already
-	// exported, then there is no need to add or update the record.
+	// If the connection is in dying state and is already exported, then there is
+	// no need to add or update the record.
 	if flowexporter.IsConnectionDying(conn) && conn.DoneExport {
 		return nil
 	}
@@ -63,8 +63,18 @@ func (fr *FlowRecords) AddOrUpdateFlowRecord(key flowexporter.ConnectionKey, con
 			IsIPv6:             isIPv6,
 			LastExportTime:     conn.StartTime,
 			IsActive:           true,
+			DoneExport:         false,
 		}
 	} else {
+		// If the connection is in dying state and the corresponding flow records are already
+		// exported, then update the DoneExport flag on the connection.
+		if flowexporter.IsConnectionDying(conn) && record.DoneExport {
+			// It is safe to update the connection as we hold the connection map
+			// lock when calling this function.
+			conn.DoneExport = true
+			delete(fr.recordsMap, key)
+			return nil
+		}
 		// set IsActive flag to true when there are changes either in stats or TCP state
 		if (conn.OriginalPackets > record.PrevPackets) || (conn.ReversePackets > record.PrevReversePackets) || record.Conn.TCPState != conn.TCPState {
 			record.IsActive = true

pkg/agent/flowexporter/types.go

@@ -88,5 +88,6 @@ type FlowRecord struct {
 	PrevReverseBytes   uint64
 	IsIPv6             bool
 	LastExportTime     time.Time
+	DoneExport         bool
 	IsActive           bool
 }