usm: Prevent negative latency (#41263)

### What does this PR do?

Prevents an edge case of negative latency.

### Motivation

Seen in staging. While we keep investigating the issue, we should at least drop the traffic.

### Describe how you validated your changes

### Additional Notes

Author: guyarb

pkg/network/protocols/http/model_linux.go

@@ -12,8 +12,8 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/DataDog/datadog-agent/pkg/network/ebpf"
 	"github.com/DataDog/datadog-agent/pkg/network/protocols"
-
 	"github.com/DataDog/datadog-agent/pkg/network/types"
 )
 
@@ -27,7 +27,10 @@ func (e *EbpfEvent) Path(buffer []byte) ([]byte, bool) {
 
 // RequestLatency returns the latency of the request in nanoseconds
 func (e *EbpfEvent) RequestLatency() float64 {
-	if uint64(e.Http.Request_started) == 0 || uint64(e.Http.Response_last_seen) == 0 {
+	if e.Http.Request_started == 0 || e.Http.Response_last_seen == 0 {
+		return 0
+	}
+	if e.Http.Response_last_seen < e.Http.Request_started {
 		return 0
 	}
 	return protocols.NSTimestampToFloat(e.Http.Response_last_seen - e.Http.Request_started)
@@ -92,7 +95,7 @@ func (e *EbpfEvent) StaticTags() uint64 {
 	return e.Http.Tags
 }
 
-// DynamicTags returns the dynamic tags associated to the HTTP trasnaction
+// DynamicTags returns the dynamic tags associated to the HTTP transaction
 func (e *EbpfEvent) DynamicTags() []string {
 	return nil
 }
@@ -101,8 +104,12 @@ func (e *EbpfEvent) DynamicTags() []string {
 func (e *EbpfEvent) String() string {
 	var output strings.Builder
 	output.WriteString("ebpfTx{")
+	output.WriteString("Conn Tuple: " + ebpf.ConnTuple(e.Tuple).String() + "', ")
 	output.WriteString("Method: '" + Method(e.Http.Request_method).String() + "', ")
 	output.WriteString("Tags: '0x" + strconv.FormatUint(e.Http.Tags, 16) + "', ")
+	output.WriteString("Request Start: " + strconv.FormatUint(e.Http.Request_started, 10) + "', ")
+	output.WriteString("Response Last Seen: " + strconv.FormatUint(e.Http.Response_last_seen, 10) + "', ")
+	output.WriteString("Latency: " + strconv.FormatFloat(e.RequestLatency(), 'f', -1, 64) + "', ")
 	output.WriteString("Fragment: '" + hex.EncodeToString(e.Http.Request_fragment[:]) + "', ")
 	output.WriteString("}")
 	return output.String()

pkg/network/protocols/http/statkeeper.go

@@ -171,7 +171,7 @@ func (h *StatKeeper) add(tx Transaction) {
 	if latency <= 0 {
 		h.telemetry.invalidLatency.Add(1)
 		if h.oversizedLogLimit.ShouldLog() {
-			log.Warnf("latency should never be equal to 0: %s", tx.String())
+			log.Warnf("latency should never be non positive: %s", tx.String())
 		}
 		return
 	}

pkg/network/protocols/http2/model_linux.go

@@ -145,7 +145,10 @@ func (tx *EbpfTx) Path(buffer []byte) ([]byte, bool) {
 
 // RequestLatency returns the latency of the request in nanoseconds
 func (tx *EbpfTx) RequestLatency() float64 {
-	if uint64(tx.Stream.Request_started) == 0 || uint64(tx.Stream.Response_last_seen) == 0 {
+	if tx.Stream.Request_started == 0 || tx.Stream.Response_last_seen == 0 {
+		return 0
+	}
+	if tx.Stream.Response_last_seen < tx.Stream.Request_started {
 		return 0
 	}
 	return protocols.NSTimestampToFloat(tx.Stream.Response_last_seen - tx.Stream.Request_started)

pkg/network/protocols/kafka/model_linux.go

@@ -48,7 +48,10 @@ func (tx *EbpfTx) ErrorCode() int8 {
 
 // RequestLatency returns the latency of the request in nanoseconds
 func (tx *EbpfTx) RequestLatency() float64 {
-	if uint64(tx.Transaction.Request_started) == 0 || uint64(tx.Transaction.Response_last_seen) == 0 {
+	if tx.Transaction.Request_started == 0 || tx.Transaction.Response_last_seen == 0 {
+		return 0
+	}
+	if tx.Transaction.Response_last_seen < tx.Transaction.Request_started {
 		return 0
 	}
 	return protocols.NSTimestampToFloat(tx.Transaction.Response_last_seen - tx.Transaction.Request_started)

pkg/network/protocols/postgres/model_linux.go

@@ -134,7 +134,10 @@ func (e *EventWrapper) Parameters() string {
 
 // RequestLatency returns the latency of the request in nanoseconds
 func (e *EventWrapper) RequestLatency() float64 {
-	if uint64(e.Tx.Request_started) == 0 || uint64(e.Tx.Response_last_seen) == 0 {
+	if e.Tx.Request_started == 0 || e.Tx.Response_last_seen == 0 {
+		return 0
+	}
+	if e.Tx.Response_last_seen < e.Tx.Request_started {
 		return 0
 	}
 	return protocols.NSTimestampToFloat(e.Tx.Response_last_seen - e.Tx.Request_started)

pkg/network/protocols/redis/model_linux.go

@@ -76,7 +76,10 @@ func (e *EventWrapper) CommandType() CommandType {
 
 // RequestLatency returns the latency of the request in nanoseconds
 func (e *EventWrapper) RequestLatency() float64 {
-	if uint64(e.Tx.Request_started) == 0 || uint64(e.Tx.Response_last_seen) == 0 {
+	if e.Tx.Request_started == 0 || e.Tx.Response_last_seen == 0 {
+		return 0
+	}
+	if e.Tx.Response_last_seen < e.Tx.Request_started {
 		return 0
 	}
 	return protocols.NSTimestampToFloat(e.Tx.Response_last_seen - e.Tx.Request_started)