Fix throughput/octetDeltaCount issue in flow visualization

build/yamls/elk-flow-collector/elk-flow-collector.yml

@@ -165,7 +165,7 @@ spec:
             - name: ELASTICSEARCH_URL
               value: "http://elasticsearch:9200"
             - name: KIBANA_DEFAULTAPPID
-              value: "dashboard/653cf1e0-2fd2-11e7-99ed-49759aed30f5"
+              value: "dashboard/3b331b30-b987-11ea-b16e-fb06687c3589"
             - name: LOGGING_QUIET
               value: "true"
           ports:

build/yamls/elk-flow-collector/logstash/filter.rb

@@ -2,7 +2,6 @@
 # register accepts the hashmap passed to "script_params"
 # it runs once at startup
 def register(params)
-    @interval = params["interval"]
     @@time_map = Hash.new
 end
 
@@ -93,9 +92,12 @@ def filter(event)
         event.set("[ipfix][reverseThroughput]", event.get("[ipfix][reverseOctetDeltaCountFromSourceNode]").to_i / duration.to_i)
         @@time_map[key] = t
      else
-        @@time_map[key] = DateTime.strptime(event.get("[ipfix][flowEndSeconds]").to_s, '%Y-%m-%dT%H:%M:%S').to_time.to_i
-        event.set("[ipfix][throughput]", event.get("[ipfix][octetDeltaCountFromSourceNode]").to_i / @interval.to_i)
-        event.set("[ipfix][reverseThroughput]", event.get("[ipfix][reverseOctetDeltaCountFromSourceNode]").to_i / @interval.to_i)
+        startTime = DateTime.strptime(event.get("[ipfix][flowStartSeconds]").to_s, '%Y-%m-%dT%H:%M:%S').to_time.to_i
+        endTime = DateTime.strptime(event.get("[ipfix][flowEndSeconds]").to_s, '%Y-%m-%dT%H:%M:%S').to_time.to_i
+        duration = endTime-startTime
+        event.set("[ipfix][throughput]", event.get("[ipfix][octetDeltaCountFromSourceNode]").to_i / duration.to_i)
+        event.set("[ipfix][reverseThroughput]", event.get("[ipfix][reverseOctetDeltaCountFromSourceNode]").to_i / duration.to_i)
+        @@time_map[key] = endTime
      end
     return [event]
 end

build/yamls/elk-flow-collector/logstash/logstash.conf

@@ -23,7 +23,6 @@ input {
 filter {
     ruby {
         path => "/usr/share/logstash/config/filter.rb"
-        script_params => { "interval" => 60 }
       }
 }
 

docs/network-flow-visibility.md

@@ -389,13 +389,6 @@ or
 svn export https://github.com/vmware-tanzu/antrea/trunk/build/yamls/elk-flow-collector/
 ```
 
-To configure the export interval as `flowExportInterval` in [Configuration](#configuration),
-modify the `interval` value (in seconds) in `elk-flow-collector/logstash/logstash.conf`
-
-```conf
-script_params => { "interval" => 60 }
-```
-
 To create the required K8s resources in the `elk-flow-collector` folder and get
 everything up-and-running, run following commands:
 

pkg/agent/flowexporter/connections/connections.go

@@ -120,9 +120,9 @@ func (cs *connectionStore) addOrUpdateConn(conn *flowexporter.Connection) {
 	existingConn, exists := cs.connections[connKey]
 
 	if exists {
+		existingConn.IsPresent = true
 		// avoid updating stats of the existing connection that is about to close
 		if flowexporter.IsConnectionDying(existingConn) {
-			existingConn.IsPresent = true
 			return
 		}
 		// Update the necessary fields that are used in generating flow records.
@@ -133,7 +133,6 @@ func (cs *connectionStore) addOrUpdateConn(conn *flowexporter.Connection) {
 		existingConn.ReverseBytes = conn.ReverseBytes
 		existingConn.ReversePackets = conn.ReversePackets
 		existingConn.TCPState = conn.TCPState
-		existingConn.IsPresent = true
 		klog.V(4).Infof("Antrea flow updated: %v", existingConn)
 	} else {
 		// sourceIP/destinationIP are mapped only to local pods and not remote pods.

pkg/agent/flowexporter/exporter/exporter.go

@@ -424,19 +424,13 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 		case "octetTotalCount":
 			ie.Value = record.Conn.OriginalBytes
 		case "packetDeltaCount":
-			deltaPkts := int64(0)
-			if record.PrevPackets != 0 {
-				deltaPkts = int64(record.Conn.OriginalPackets) - int64(record.PrevPackets)
-			}
+			deltaPkts := int64(record.Conn.OriginalPackets) - int64(record.PrevPackets)
 			if deltaPkts < 0 {
 				klog.Warningf("Packet delta count for connection should not be negative: %d", deltaPkts)
 			}
 			ie.Value = uint64(deltaPkts)
 		case "octetDeltaCount":
-			deltaBytes := int64(0)
-			if record.PrevBytes != 0 {
-				deltaBytes = int64(record.Conn.OriginalBytes) - int64(record.PrevBytes)
-			}
+			deltaBytes := int64(record.Conn.OriginalBytes) - int64(record.PrevBytes)
 			if deltaBytes < 0 {
 				klog.Warningf("Byte delta count for connection should not be negative: %d", deltaBytes)
 			}
@@ -446,19 +440,13 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 		case "reverseOctetTotalCount":
 			ie.Value = record.Conn.ReverseBytes
 		case "reversePacketDeltaCount":
-			deltaPkts := int64(0)
-			if record.PrevReversePackets != 0 {
-				deltaPkts = int64(record.Conn.ReversePackets) - int64(record.PrevReversePackets)
-			}
+			deltaPkts := int64(record.Conn.ReversePackets) - int64(record.PrevReversePackets)
 			if deltaPkts < 0 {
 				klog.Warningf("Packet delta count for connection should not be negative: %d", deltaPkts)
 			}
 			ie.Value = uint64(deltaPkts)
 		case "reverseOctetDeltaCount":
-			deltaBytes := int64(0)
-			if record.PrevReverseBytes != 0 {
-				deltaBytes = int64(record.Conn.ReverseBytes) - int64(record.PrevReverseBytes)
-			}
+			deltaBytes := int64(record.Conn.ReverseBytes) - int64(record.PrevReverseBytes)
 			if deltaBytes < 0 {
 				klog.Warningf("Byte delta count for connection should not be negative: %d", deltaBytes)
 			}