Log connection lifetime in collector

collector/collector.go

@@ -244,13 +244,14 @@ func (c *Collector) collect() error {
 		} else {
 			c.podMetrics[groupKey] = &rawNetworkMetric{
 				RawNetworkMetric: &pb.RawNetworkMetric{
-					SrcIp:      dns.ToIPint32(conn.Src.IP()),
-					DstIp:      dns.ToIPint32(conn.Dst.IP()),
-					TxBytes:    int64(conn.TxBytes),
-					TxPackets:  int64(conn.TxPackets),
-					RxBytes:    int64(conn.RxBytes),
-					RxPackets:  int64(conn.RxPackets),
-					Proto:      int32(conn.Proto),
+					SrcIp:     dns.ToIPint32(conn.Src.IP()),
+					DstIp:     dns.ToIPint32(conn.Dst.IP()),
+					TxBytes:   int64(conn.TxBytes),
+					TxPackets: int64(conn.TxPackets),
+					RxBytes:   int64(conn.RxBytes),
+					RxPackets: int64(conn.RxPackets),
+					Proto:     int32(conn.Proto),
+					Lifetime:  conn.Lifetime.Format(time.RFC3339),
 				},
 				lifetime: conn.Lifetime,
 			}

collector/collector_test.go

@@ -122,6 +122,7 @@ func TestCollector(t *testing.T) {
 			RxBytes:   30,
 			RxPackets: 1,
 			Proto:     6,
+			Lifetime:  "0001-01-01T00:00:00Z",
 		}, items[0])
 
 		r.Equal(&pb.RawNetworkMetric{
@@ -132,6 +133,7 @@ func TestCollector(t *testing.T) {
 			RxBytes:   0,
 			RxPackets: 0,
 			Proto:     17,
+			Lifetime:  "0001-01-01T00:00:00Z",
 		}, items[1])
 	})
 
@@ -247,6 +249,7 @@ func TestCollector(t *testing.T) {
 			RxBytes:   0,
 			RxPackets: 0,
 			Proto:     6,
+			Lifetime:  "0001-01-01T00:00:00Z",
 		})
 
 		r.Contains(items, &pb.RawNetworkMetric{
@@ -257,6 +260,7 @@ func TestCollector(t *testing.T) {
 			RxBytes:   0,
 			RxPackets: 0,
 			Proto:     6,
+			Lifetime:  "0001-01-01T00:00:00Z",
 		})
 	})
 

exporter/exporter.go

@@ -156,6 +156,7 @@ func (e *Exporter) export(ctx context.Context) error {
 		for _, rawMetrics := range batch.Items {
 			podMetrics, err := e.buildPodNetworkMetric(rawMetrics)
 			if err != nil {
+				e.log.Infof("processing metrics: %d %d %s", rawMetrics.SrcIp, rawMetrics.DstIp, rawMetrics.Lifetime)
 				if errors.Is(err, kube.ErrNotFound) {
 					e.log.Warnf("skipping pod metrics: %v", err)
 				} else {

kube/watcher_test.go

@@ -95,23 +95,7 @@ func TestWatcher(t *testing.T) {
 		},
 	}
 
-	// pod exited a while ago, and should not be found
-	p4 := &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:              "p4",
-			Namespace:         "team1",
-			CreationTimestamp: thirtySecsAgo,
-		},
-		Spec: corev1.PodSpec{
-			NodeName: n1.Name,
-		},
-		Status: corev1.PodStatus{
-			PodIP: "10.14.7.13",
-			Phase: corev1.PodSucceeded,
-		},
-	}
-
-	clientset := fake.NewSimpleClientset(n1, p1, p2, p3, p4)
+	clientset := fake.NewSimpleClientset(n1, p1, p2, p3)
 
 	informersFactory := informers.NewSharedInformerFactoryWithOptions(clientset, 30*time.Second)
 	podsInformer := informersFactory.Core().V1().Pods().Informer()

pb/metrics.proto

@@ -10,6 +10,7 @@ message RawNetworkMetric {
   int64 rx_bytes = 5;
   int64 rx_packets = 6;
   int32 proto = 7;
+  string lifetime = 8;
 }
 
 message RawNetworkMetricBatch {