import flow package in init.go to enable flow probe

cmd/exporter/init.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	_ "github.com/alibaba/kubeskoop/pkg/exporter/probe/flow"
 	_ "github.com/alibaba/kubeskoop/pkg/exporter/probe/nlconntrack"
 	_ "github.com/alibaba/kubeskoop/pkg/exporter/probe/nlqdisc"
 	_ "github.com/alibaba/kubeskoop/pkg/exporter/probe/procfd"

pkg/exporter/probe/flow/flow.go

@@ -116,17 +116,32 @@ func (p *metricsProbe) collectOnce(emit probe.Emit) error {
 			val.Bytes += values[i].Bytes
 			val.Packets += values[i].Packets
 		}
-		emit("bytes", []string{}, float64(val.Bytes))
-		emit("packets", []string{}, float64(val.Packets))
 
-		fmt.Printf("proto: %d %s:%d->%s:%d pkts: %d, bytes: %d\n",
-			key.Proto,
+		var protocol string
+
+		switch key.Proto {
+		case 1:
+			protocol = "icmp"
+		case 6:
+			protocol = "tcp"
+		case 17:
+			protocol = "udp"
+		case 132:
+			protocol = "sctp"
+		default:
+			log.Errorf("%s unknown ip protocol number %d", probeName, key.Proto)
+		}
+
+		labels := []string{
+			protocol,
 			toIPString(key.Src),
-			htons(key.Sport),
 			toIPString(key.Dst),
-			htons(key.Dport),
-			val.Packets,
-			val.Bytes)
+			fmt.Sprintf("%d", htons(key.Sport)),
+			fmt.Sprintf("%d", htons(key.Dport)),
+		}
+
+		emit("bytes", labels, float64(val.Bytes))
+		emit("packets", labels, float64(val.Packets))
 	}
 	return nil
 }

pkg/exporter/probe/metric.go → pkg/exporter/probe/metrics.go

@@ -114,7 +114,12 @@ func (b *BatchMetrics) Collect(metrics chan<- prometheus.Metric) {
 			log.Errorf("%s undeclared metrics %s", b.name, name)
 			return
 		}
-		metrics <- prometheus.MustNewConstMetric(info.desc, info.valueType, val, labels...)
+		m, err := prometheus.NewConstMetric(info.desc, info.valueType, val, labels...)
+		if err != nil {
+			log.Errorf("%s failed create metrics, err: %v", b.name, err)
+			return
+		}
+		metrics <- m
 	}
 
 	err := b.ProbeCollector(emit)