Fix zero delta count for first record

IsPresent field of connection should be updated before having
IsConnectionDying check otherwise IsConnectionDying will always
return true, which makes existing connections cannot be updated and
octetDeltaCount always return 0.

This commit also changes the delta count of first record from zero
to its total delta count, modifies throughput calculation of first
record in logstash config and changes names of thoughput diagram
from 'throughput' to 'cumulative bytes'.

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)
 			}