Something went wrong. Try again.
router flow monitor
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182package export
import ( "context" "errors" "fmt" "net/netip" "os" "strings" "testing" "time"
"github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" "ysun.co/rfm/collector" "ysun.co/rfm/config")
// --- mock types ---
type mockIfaceStats struct { entries []IfaceStatsEntry err error rate uint32 rateErr error}
func (m *mockIfaceStats) IfaceStats() ([]IfaceStatsEntry, error) { return m.entries, m.err}
func (m *mockIfaceStats) SampleRate() (uint32, error) { if m.rateErr != nil { return 0, m.rateErr } if m.rate == 0 { return 1, nil } return m.rate, nil}
type staticEnricher struct { srcASN uint32 srcCity string dstASN uint32 dstCity string}
func (e *staticEnricher) Enrich(src, dst netip.Addr) (collector.Labels, collector.Labels) { return collector.Labels{ASN: e.srcASN, City: e.srcCity}, collector.Labels{ASN: e.dstASN, City: e.dstCity}}
type exportMockReader struct { events [][]byte idx int dropErr error}
func (m *exportMockReader) ReadRawEvent() ([]byte, error) { if m.idx >= len(m.events) { return nil, os.ErrDeadlineExceeded } raw := m.events[m.idx] m.idx++ return raw, nil}
func (m *exportMockReader) SetDeadline(t time.Time) {}
func (m *exportMockReader) DroppedEvents() (uint64, error) { return 0, m.dropErr}
func (m *exportMockReader) Close() error { return nil }
// --- test helpers ---
// collectAll calls Collect and returns a map of metric name -> summed valuefunc collectAll(t *testing.T, mc *MetricsCollector) map[string]float64 { t.Helper() metrics := collectMetrics(t, mc) result := make(map[string]float64) for _, m := range metrics { d := new(dto.Metric) if err := m.Write(d); err != nil { t.Fatalf("write metric: %v", err) } name := extractName(m.Desc()) var val float64 switch { case d.Counter != nil: val = d.Counter.GetValue() case d.Gauge != nil: val = d.Gauge.GetValue() } result[name] += val } return result}
// collectMetrics calls Collect and returns all emitted metricsfunc collectMetrics(t *testing.T, mc *MetricsCollector) []prometheus.Metric { t.Helper() ch := make(chan prometheus.Metric, 100) mc.Collect(ch) close(ch) var out []prometheus.Metric for m := range ch { out = append(out, m) } return out}
// extractName parses fqName from Desc.String() format: Desc{fqName: "rfm_...", ...}func extractName(desc *prometheus.Desc) string { s := desc.String() const prefix = "fqName: \"" i := strings.Index(s, prefix) if i < 0 { return "" } s = s[i+len(prefix):] before, _, ok := strings.Cut(s, "\"") if !ok { return "" } return before}
// assertCounter checks that the named metric has the expected valuefunc assertCounter(t *testing.T, vals map[string]float64, name string, want float64) { t.Helper() got, ok := vals[name] if !ok { t.Errorf("metric %q not found", name) return } if got != want { t.Errorf("%s = %g, want %g", name, got, want) }}
// assertGauge checks that the named metric has the expected valuefunc assertGauge(t *testing.T, vals map[string]float64, name string, want float64) { t.Helper() got, ok := vals[name] if !ok { t.Errorf("metric %q not found", name) return } if got != want { t.Errorf("%s = %g, want %g", name, got, want) }}
// metricLabels returns the label name->value map for a metricfunc metricLabels(t *testing.T, m prometheus.Metric) map[string]string { t.Helper() d := new(dto.Metric) if err := m.Write(d); err != nil { t.Fatalf("write metric: %v", err) } labels := make(map[string]string) for _, lp := range d.Label { labels[lp.GetName()] = lp.GetValue() } return labels}
// metricValue returns the numeric value of a metric (counter or gauge)func metricValue(t *testing.T, m prometheus.Metric) float64 { t.Helper() d := new(dto.Metric) if err := m.Write(d); err != nil { t.Fatalf("write metric: %v", err) } switch { case d.Counter != nil: return d.Counter.GetValue() case d.Gauge != nil: return d.Gauge.GetValue() } return 0}
// --- tests ---
func TestMetricsCollectorImplementsInterface(t *testing.T) { var _ prometheus.Collector = (*MetricsCollector)(nil)}
func TestDescribe(t *testing.T) { mc := New(nil, nil) ch := make(chan *prometheus.Desc, 32) mc.Describe(ch) close(ch)
var descs []*prometheus.Desc for d := range ch { descs = append(descs, d) }
if got, want := len(descs), len(allDescs); got != want { t.Fatalf("got %d descriptors, want %d", got, want) }
names := make(map[string]bool) for _, d := range descs { names[extractName(d)] = true }
want := []string{ "rfm_interface_rx_bytes_total", "rfm_interface_tx_bytes_total", "rfm_interface_rx_packets_total", "rfm_interface_tx_packets_total", "rfm_flow_bytes", "rfm_flow_packets", "rfm_flow_sampled_bytes", "rfm_flow_sampled_packets", "rfm_flow_bytes_total", "rfm_flow_packets_total", "rfm_flow_sampled_bytes_total", "rfm_flow_sampled_packets_total", "rfm_collector_active_flows", "rfm_collector_dropped_events_total", "rfm_collector_forced_evictions_total", "rfm_collector_folded_flows_total", "rfm_errors_total", "rfm_bpf_sample_rate", "rfm_ipfix_connected", "rfm_ipfix_dials_total", "rfm_ipfix_dial_errors_total", "rfm_ipfix_messages_total", "rfm_ipfix_records_total", "rfm_ipfix_dropped_records_total", "rfm_ipfix_send_errors_total", } for _, w := range want { if !names[w] { t.Errorf("missing descriptor %q", w) } }}
func TestCollectHealth(t *testing.T) { c := collector.New(time.Minute, nil, config.DefaultMaxFlows) c.Record(collector.FlowEvent{ Ifindex: 1, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("10.0.0.1"), DstAddr: netip.MustParseAddr("10.0.0.2"), SrcPort: 12345, DstPort: 80, Segs: 1, Len: 100, }, time.Now())
mc := New(nil, c) vals := collectAll(t, mc)
assertGauge(t, vals, "rfm_collector_active_flows", 1) assertCounter(t, vals, "rfm_collector_dropped_events_total", 0) assertCounter(t, vals, "rfm_collector_forced_evictions_total", 0)}
func TestCollectIfaceStats(t *testing.T) { src := &mockIfaceStats{ entries: []IfaceStatsEntry{ {Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000}, // rx ipv4 {Ifindex: 1, Dir: 1, Proto: 4, Packets: 5, Bytes: 500}, // tx ipv4 {Ifindex: 1, Dir: 0, Proto: 6, Packets: 3, Bytes: 300}, // rx ipv6 }, }
mc := New(src, nil) vals := collectAll(t, mc)
assertCounter(t, vals, "rfm_interface_rx_bytes_total", 1300) // 1000 + 300 assertCounter(t, vals, "rfm_interface_tx_bytes_total", 500) assertCounter(t, vals, "rfm_interface_rx_packets_total", 13) // 10 + 3 assertCounter(t, vals, "rfm_interface_tx_packets_total", 5)}
func TestCollectIfaceStatsSkipsKeysSeenTwice(t *testing.T) { // deleting the key under the cursor restarts a hash map walk, so a scrape // that overlaps a detach can read a key twice src := &mockIfaceStats{ entries: []IfaceStatsEntry{ {Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000}, {Ifindex: 1, Dir: 1, Proto: 4, Packets: 5, Bytes: 500}, {Ifindex: 1, Dir: 0, Proto: 4, Packets: 11, Bytes: 1100}, }, } mc := New(src, nil) mc.resolveIfname = func(uint32) string { return "eth0" } reg := prometheus.NewRegistry() reg.MustRegister(mc)
mfs, err := reg.Gather() if err != nil { t.Fatalf("gather: %v", err) } for _, mf := range mfs { if mf.GetName() != "rfm_interface_rx_bytes_total" { continue } if got := len(mf.GetMetric()); got != 1 { t.Fatalf("rx series = %d, want 1", got) } if got := mf.GetMetric()[0].GetCounter().GetValue(); got != 1100 { t.Fatalf("rx bytes = %v, want the later read 1100", got) } return } t.Fatal("rfm_interface_rx_bytes_total missing")}
func TestScrapeSurvivesAnInterfaceNameThatIsNotUTF8(t *testing.T) { // a link name is any bytes to the kernel, the registry refuses the whole // scrape over one label that is not utf-8 src := &mockIfaceStats{ entries: []IfaceStatsEntry{{Ifindex: 7, Dir: 0, Proto: 4, Packets: 1, Bytes: 60}}, } c := collector.New(time.Minute, nil, 16) c.Record(collector.FlowEvent{ Ifindex: 7, Proto: 17, SrcPort: 5000, DstPort: 53, SrcAddr: netip.MustParseAddr("10.0.0.1"), DstAddr: netip.MustParseAddr("10.0.0.2"), Segs: 1, Len: 60, }, time.Now()) mc := New(src, c) mc.resolveIfname = func(uint32) string { return "d\xff" } reg := prometheus.NewRegistry() reg.MustRegister(mc)
mfs, err := reg.Gather() if err != nil { t.Fatalf("gather: %v", err) } // the interface counters and the flow series name the link alike seen := map[string]bool{} for _, mf := range mfs { for _, m := range mf.GetMetric() { for _, l := range m.GetLabel() { if l.GetName() != "ifname" { continue } if l.GetValue() != "d\uFFFD" { t.Fatalf("%s ifname = %q, want the invalid byte replaced", mf.GetName(), l.GetValue()) } seen[mf.GetName()] = true } } } for _, name := range []string{"rfm_interface_rx_packets_total", "rfm_flow_packets_total"} { if !seen[name] { t.Fatalf("%s missing", name) } }}
func TestCollectCachesIfnames(t *testing.T) { src := &mockIfaceStats{ entries: []IfaceStatsEntry{ {Ifindex: 7, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000}, {Ifindex: 7, Dir: 1, Proto: 6, Packets: 5, Bytes: 500}, }, }
c := collector.New(time.Minute, nil, config.DefaultMaxFlows) now := time.Now() c.Record(collector.FlowEvent{ Ifindex: 7, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("10.0.0.1"), DstAddr: netip.MustParseAddr("10.0.0.2"), SrcPort: 1000, DstPort: 80, Segs: 1, Len: 100, }, now) c.Record(collector.FlowEvent{ Ifindex: 7, Dir: 1, Proto: 17, SrcAddr: netip.MustParseAddr("10.0.0.3"), DstAddr: netip.MustParseAddr("10.0.0.4"), SrcPort: 2000, DstPort: 53, Segs: 1, Len: 200, }, now)
mc := New(src, c) calls := make(map[uint32]int) mc.resolveIfname = func(ifindex uint32) string { calls[ifindex]++ return fmt.Sprintf("if-%d", ifindex) }
// one resolve per scrape however many series mention the interface, // a second scrape resolves again so a renamed interface shows up collectMetrics(t, mc) if got := calls[7]; got != 1 { t.Fatalf("ifname resolver calls for ifindex 7 = %d, want 1", got) } collectMetrics(t, mc) if got := calls[7]; got != 2 { t.Fatalf("ifname resolver calls after two scrapes = %d, want 2", got) }}
func TestCollectFlowsNoEnricher(t *testing.T) { c := collector.New(time.Minute, nil, config.DefaultMaxFlows) c.Record(collector.FlowEvent{ Ifindex: 2, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("192.168.1.1"), DstAddr: netip.MustParseAddr("192.168.1.2"), SrcPort: 1234, DstPort: 443, Segs: 1, Len: 200, }, time.Now())
mc := New(nil, c) vals := collectAll(t, mc)
assertGauge(t, vals, "rfm_flow_bytes", 200) assertGauge(t, vals, "rfm_flow_packets", 1) assertGauge(t, vals, "rfm_flow_sampled_bytes", 200) assertGauge(t, vals, "rfm_flow_sampled_packets", 1)
// check labels, enrichment should be empty strings metrics := collectMetrics(t, mc) for _, m := range metrics { name := extractName(m.Desc()) if name != "rfm_flow_bytes" { continue } labels := metricLabels(t, m) if labels["src_asn"] != "" { t.Errorf("src_asn = %q, want empty", labels["src_asn"]) } if labels["dst_asn"] != "" { t.Errorf("dst_asn = %q, want empty", labels["dst_asn"]) } if labels["direction"] != "ingress" { t.Errorf("direction = %q, want ingress", labels["direction"]) } }}
func TestCollectFlowsWithEnricher(t *testing.T) { e := &staticEnricher{ srcASN: 64512, srcCity: "Berlin", dstASN: 13335, dstCity: "London", } c := collector.New(time.Minute, e, config.DefaultMaxFlows) c.Record(collector.FlowEvent{ Ifindex: 3, Dir: 1, Proto: 17, SrcAddr: netip.MustParseAddr("10.1.0.1"), DstAddr: netip.MustParseAddr("10.2.0.1"), SrcPort: 5000, DstPort: 53, Segs: 1, Len: 64, }, time.Now())
mc := New(nil, c) metrics := collectMetrics(t, mc)
var found bool for _, m := range metrics { name := extractName(m.Desc()) if name != "rfm_flow_bytes" { continue } found = true labels := metricLabels(t, m)
if labels["src_asn"] != "64512" { t.Errorf("src_asn = %q, want 64512", labels["src_asn"]) } if labels["dst_asn"] != "13335" { t.Errorf("dst_asn = %q, want 13335", labels["dst_asn"]) } if labels["src_city"] != "Berlin" { t.Errorf("src_city = %q, want Berlin", labels["src_city"]) } if labels["dst_city"] != "London" { t.Errorf("dst_city = %q, want London", labels["dst_city"]) } if labels["direction"] != "egress" { t.Errorf("direction = %q, want egress", labels["direction"]) }
val := metricValue(t, m) if val != 64 { t.Errorf("flow_bytes = %g, want 64", val) } } if !found { t.Error("rfm_flow_bytes metric not found") }}
func TestCollectFlowsScalesBySampleRate(t *testing.T) { src := &mockIfaceStats{rate: 10}
c := collector.New(time.Minute, nil, config.DefaultMaxFlows) c.SetSampleRate(10, 0) c.Record(collector.FlowEvent{ Ifindex: 2, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("192.168.1.1"), DstAddr: netip.MustParseAddr("192.168.1.2"), SrcPort: 1234, DstPort: 443, Segs: 1, Len: 200, }, time.Now())
mc := New(src, c) vals := collectAll(t, mc)
assertGauge(t, vals, "rfm_flow_sampled_bytes", 200) assertGauge(t, vals, "rfm_flow_sampled_packets", 1) assertGauge(t, vals, "rfm_flow_bytes", 2000) assertGauge(t, vals, "rfm_flow_packets", 10) assertGauge(t, vals, "rfm_bpf_sample_rate", 10)}
func TestCollectNilSources(t *testing.T) { mc := New(nil, nil) // must not panic vals := collectAll(t, mc) // should have no iface or flow metrics, but also no panic _ = vals}
func TestCollectFlowsAggregatesDuplicateLabels(t *testing.T) { // two flows with different ports but same exported labels // must produce one aggregated metric, not duplicate series c := collector.New(time.Minute, nil, config.DefaultMaxFlows) now := time.Now() c.Record(collector.FlowEvent{ Ifindex: 1, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("10.0.0.1"), DstAddr: netip.MustParseAddr("10.0.0.2"), SrcPort: 1000, DstPort: 80, Segs: 1, Len: 100, }, now) c.Record(collector.FlowEvent{ Ifindex: 1, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("10.0.0.1"), DstAddr: netip.MustParseAddr("10.0.0.2"), SrcPort: 2000, DstPort: 80, Segs: 1, Len: 200, }, now)
mc := New(nil, c)
// use a real registry to verify no duplicate series reg := prometheus.NewRegistry() reg.MustRegister(mc) mfs, err := reg.Gather() if err != nil { t.Fatalf("gather: %v", err) }
// find rfm_flow_bytes and check it has exactly one series for _, mf := range mfs { if mf.GetName() != "rfm_flow_bytes" && mf.GetName() != "rfm_flow_sampled_bytes" { continue } if got := len(mf.GetMetric()); got != 1 { t.Fatalf("%s series count = %d, want 1 (aggregated)", mf.GetName(), got) } val := mf.GetMetric()[0].GetGauge().GetValue() if val != 300 { t.Fatalf("%s = %g, want 300 (100+200)", mf.GetName(), val) } }}
func TestCollectIfaceStatsError(t *testing.T) { src := &mockIfaceStats{ entries: []IfaceStatsEntry{ {Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000}, }, err: fmt.Errorf("map iteration failed"), }
mc := New(src, nil) metrics := collectMetrics(t, mc)
// on error, no iface metrics should be emitted for _, m := range metrics { name := extractName(m.Desc()) if strings.HasPrefix(name, "rfm_interface_") { t.Errorf("unexpected iface metric %q on error", name) } }}
func TestCollectIfaceStatsFamily(t *testing.T) { // iface stats proto=4 is IPv4, proto=6 is IPv6 // the label should be "family" with values "ipv4" / "ipv6" src := &mockIfaceStats{ entries: []IfaceStatsEntry{ {Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000}, {Ifindex: 1, Dir: 0, Proto: 6, Packets: 5, Bytes: 500}, }, }
mc := New(src, nil) metrics := collectMetrics(t, mc)
families := make(map[string]bool) for _, m := range metrics { name := extractName(m.Desc()) if name != "rfm_interface_rx_bytes_total" { continue } labels := metricLabels(t, m) families[labels["family"]] = true }
if !families["ipv4"] { t.Error("expected family=ipv4 label") } if !families["ipv6"] { t.Error("expected family=ipv6 label") }}
func TestCollectErrorsTotal(t *testing.T) { src := &mockIfaceStats{ entries: nil, err: fmt.Errorf("map broken"), }
mc := New(src, nil) metrics := collectMetrics(t, mc)
var found bool for _, m := range metrics { name := extractName(m.Desc()) if name != "rfm_errors_total" { continue } found = true labels := metricLabels(t, m) if labels["subsystem"] != "bpf_map" { t.Errorf("subsystem = %q, want bpf_map", labels["subsystem"]) } val := metricValue(t, m) if val != 1 { t.Errorf("errors_total = %g, want 1", val) } } if !found { t.Error("rfm_errors_total not found") }}
func TestCollectErrorsSubsystemLabels(t *testing.T) { // exercise both subsystem labels via a collector that has // accumulated ring_buffer and bpf_map errors through Run // the drop counter is polled by the eviction ticker, so keep it short c := collector.New(200*time.Millisecond, nil, config.DefaultMaxFlows)
mr := &exportMockReader{ events: [][]byte{{0xde, 0xad}}, // garbage -> decode error -> ring_buffer dropErr: fmt.Errorf("map read failed"), }
ctx, cancel := context.WithCancel(context.Background()) errCh := make(chan error, 1) go func() { errCh <- c.Run(ctx, mr) }()
// wait for errors to be counted deadline := time.Now().Add(time.Second) for { s := c.Stats() if s.RingBufErrors > 0 && s.BPFMapErrors > 0 { break } if time.Now().After(deadline) { t.Fatal("timed out waiting for error counters") } time.Sleep(5 * time.Millisecond) } cancel() <-errCh
mc := New(nil, c) metrics := collectMetrics(t, mc)
subsystems := make(map[string]float64) for _, m := range metrics { if extractName(m.Desc()) != "rfm_errors_total" { continue } labels := metricLabels(t, m) subsystems[labels["subsystem"]] = metricValue(t, m) }
if _, ok := subsystems["ring_buffer"]; !ok { t.Error("missing rfm_errors_total{subsystem=\"ring_buffer\"}") } else if subsystems["ring_buffer"] < 1 { t.Errorf("ring_buffer errors = %g, want >= 1", subsystems["ring_buffer"]) }
if _, ok := subsystems["bpf_map"]; !ok { t.Error("missing rfm_errors_total{subsystem=\"bpf_map\"}") } else if subsystems["bpf_map"] < 1 { t.Errorf("bpf_map errors = %g, want >= 1", subsystems["bpf_map"]) }}
func TestCollectErrorsTotalZeroOnSuccess(t *testing.T) { src := &mockIfaceStats{ entries: []IfaceStatsEntry{ {Ifindex: 1, Dir: 0, Proto: 4, Packets: 1, Bytes: 100}, }, }
mc := New(src, nil) vals := collectAll(t, mc) assertCounter(t, vals, "rfm_errors_total", 0)}
func TestCollectErrorsTotalOnSampleRateLookupError(t *testing.T) { src := &mockIfaceStats{ rateErr: fmt.Errorf("config map broken"), }
c := collector.New(time.Minute, nil, config.DefaultMaxFlows) c.Record(collector.FlowEvent{ Ifindex: 2, Dir: 0, Proto: 6, SrcAddr: netip.MustParseAddr("192.168.1.1"), DstAddr: netip.MustParseAddr("192.168.1.2"), SrcPort: 1234, DstPort: 443, Segs: 1, Len: 200, }, time.Now())
mc := New(src, c) metrics := collectMetrics(t, mc)
var found bool for _, m := range metrics { // a rate the scrape could not read is left out, a stand in of 1 // would claim the programs sample every packet if extractName(m.Desc()) == "rfm_bpf_sample_rate" { t.Errorf("rfm_bpf_sample_rate = %g after a failed read, want it left out", metricValue(t, m)) } if extractName(m.Desc()) != "rfm_errors_total" { continue } labels := metricLabels(t, m) if labels["subsystem"] != "bpf_map" { continue } found = true if val := metricValue(t, m); val != 1 { t.Errorf("bpf_map errors = %g, want 1", val) } } if !found { t.Error("missing rfm_errors_total{subsystem=\"bpf_map\"}") }}
// refusingIfaceStats is a probe whose interface counter map refused updatestype refusingIfaceStats struct { mockIfaceStats refused uint64 err error}
func (m *refusingIfaceStats) IfaceStatsErrors() (uint64, error) { return m.refused, m.err}
// errorsBySubsystem returns the rfm_errors_total values of one scrapefunc errorsBySubsystem(t *testing.T, mc *MetricsCollector) map[string]float64 { t.Helper() got := map[string]float64{} for _, m := range collectMetrics(t, mc) { if extractName(m.Desc()) == "rfm_errors_total" { got[metricLabels(t, m)["subsystem"]] += metricValue(t, m) } } return got}
func TestCollectErrorsCountRefusedIfaceStatsUpdates(t *testing.T) { src := &refusingIfaceStats{refused: 7} mc := New(src, nil) if got := errorsBySubsystem(t, mc)["bpf_map"]; got != 7 { t.Fatalf("bpf_map errors = %v, want the 7 refused counter updates", got) }
// a failed read counts as one more error and the refused updates keep // their last value, so the counter never goes back src.err = errors.New("lookup failed") if got := errorsBySubsystem(t, mc)["bpf_map"]; got != 8 { t.Fatalf("bpf_map errors = %v after a failed read, want 8", got) }}
// the probe source reports both counts of errors the programs keepvar ( _ IfaceStatsErrorSource = (*ProbeSource)(nil) _ GSOHeaderErrorSource = (*ProbeSource)(nil))
// unparsedGSO is a probe whose programs counted gso skbs without parsing// their headerstype unparsedGSO struct { mockIfaceStats unparsed uint64 err error}
func (m *unparsedGSO) GSOHeaderErrors() (uint64, error) { return m.unparsed, m.err}
func TestCollectErrorsCountUnparsedGSOHeaders(t *testing.T) { src := &unparsedGSO{unparsed: 4} mc := New(src, nil) if got := errorsBySubsystem(t, mc); got["gso_header"] != 4 || got["bpf_map"] != 0 { t.Fatalf("errors = %v, want the 4 unparsed gso skbs as gso_header errors", got) }
// a failed read is a bpf_map error and the gso_header errors keep their // last value, so neither goes back src.err = errors.New("lookup failed") if got := errorsBySubsystem(t, mc); got["gso_header"] != 4 || got["bpf_map"] != 1 { t.Fatalf("errors = %v after a failed read, want 4 gso_header and 1 bpf_map", got) }
// a source that keeps no such count has no series for it if got := errorsBySubsystem(t, New(&mockIfaceStats{}, nil)); len(got) != 1 { t.Fatalf("errors = %v from a source without gso header errors, want bpf_map alone", got) }}
func TestCollectErrorsOfOtherSubsystems(t *testing.T) { mc := New(nil, collector.New(time.Minute, nil, config.DefaultMaxFlows)) var parse, dropped uint64 = 3, 2 mc.AddErrors("bmp", func() uint64 { return parse }) mc.AddErrors("netlink", func() uint64 { return dropped }) mc.AddErrors("bpf_map", func() uint64 { return 5 })
got := errorsBySubsystem(t, mc) want := map[string]float64{"bmp": 3, "netlink": 2, "bpf_map": 5, "ring_buffer": 0, "ipfix": 0} if len(got) != len(want) { t.Fatalf("subsystems = %v, want %v", got, want) } for s, v := range want { if got[s] != v { t.Fatalf("%s errors = %v, want %v", s, got[s], v) } }
// every subsystem is one series, a registry refuses a repeated one reg := prometheus.NewRegistry() reg.MustRegister(mc) if _, err := reg.Gather(); err != nil { t.Fatalf("gather: %v", err) }}
func TestCollectIPFIXStats(t *testing.T) { c := collector.New(30*time.Second, nil, config.DefaultMaxFlows) mc := New(nil, c) mc.SetIPFIX(func() IPFIXStats { return IPFIXStats{ Connected: true, Dials: 2, DialErrors: 1, Messages: 10, Records: 200, QueueDropped: 3, Unsent: 4, EncodeErrors: 0, SendFailed: 40, SendErrors: map[string]uint64{"EPERM": 5, "ECONNREFUSED": 1}, } })
metrics := collectMetrics(t, mc) got := map[string]float64{} for _, m := range metrics { d := new(dto.Metric) if err := m.Write(d); err != nil { t.Fatal(err) } var name strings.Builder name.WriteString(extractName(m.Desc())) for _, l := range d.Label { name.WriteString("{" + l.GetName() + "=" + l.GetValue() + "}") } switch { case d.Counter != nil: got[name.String()] = d.Counter.GetValue() case d.Gauge != nil: got[name.String()] = d.Gauge.GetValue() } }
want := map[string]float64{ "rfm_ipfix_connected": 1, "rfm_ipfix_dials_total": 2, "rfm_ipfix_dial_errors_total": 1, "rfm_ipfix_messages_total": 10, "rfm_ipfix_records_total": 200, "rfm_ipfix_dropped_records_total{reason=queue_full}": 3, "rfm_ipfix_dropped_records_total{reason=unconnected}": 4, "rfm_ipfix_dropped_records_total{reason=encode}": 0, "rfm_ipfix_dropped_records_total{reason=send}": 40, "rfm_ipfix_send_errors_total{errno=EPERM}": 5, "rfm_ipfix_send_errors_total{errno=ECONNREFUSED}": 1, // every lost record counts once by its reason, the failed messages // and dials lose nothing beyond the records counted "rfm_errors_total{subsystem=ipfix}": 47, } for name, val := range want { if got[name] != val { t.Errorf("%s = %v, want %v", name, got[name], val) } }}
func TestCollectRollupCountersSurviveEviction(t *testing.T) { e := &staticEnricher{srcASN: 64500, dstASN: 64501, srcCity: "Oslo", dstCity: "Lima"} c := collector.New(time.Second, e, config.DefaultMaxFlows) c.SetSampleRate(10, 0)
t0 := time.Now() ev := collector.FlowEvent{ Ifindex: 2, Dir: 1, Proto: 17, SrcAddr: netip.MustParseAddr("192.168.1.1"), DstAddr: netip.MustParseAddr("192.168.1.2"), SrcPort: 1234, DstPort: 53, Segs: 1, Len: 100, } c.Record(ev, t0) other := ev other.SrcPort = 4321 c.Record(other, t0)
// the first scrape shows the new tuple at zero mc := New(nil, c) if got := collectAll(t, mc)["rfm_flow_bytes_total"]; got != 0 { t.Fatalf("rfm_flow_bytes_total on the first scrape = %v, want 0", got) }
// evicting the flows resets the gauges but not the counters c.Evict(t0.Add(5 * time.Second))
vals := collectAll(t, mc) if got := vals["rfm_flow_bytes"]; got != 0 { t.Fatalf("rfm_flow_bytes = %v after eviction, want no live flows", got) } if got := vals["rfm_flow_bytes_total"]; got != 2000 { t.Fatalf("rfm_flow_bytes_total = %v, want 2000", got) } if got := vals["rfm_flow_packets_total"]; got != 20 { t.Fatalf("rfm_flow_packets_total = %v, want 20", got) } if got := vals["rfm_flow_sampled_bytes_total"]; got != 200 { t.Fatalf("rfm_flow_sampled_bytes_total = %v, want 200", got) } if got := vals["rfm_flow_sampled_packets_total"]; got != 2 { t.Fatalf("rfm_flow_sampled_packets_total = %v, want 2", got) }
for _, m := range collectMetrics(t, mc) { if extractName(m.Desc()) != "rfm_flow_bytes_total" { continue } labels := metricLabels(t, m) if labels["src_asn"] != "64500" || labels["dst_city"] != "Lima" || labels["direction"] != "egress" { t.Fatalf("unexpected labels %v", labels) } }
// the counters outlive their flows for the retention, then leave the scrape c.Evict(t0.Add(collector.RollupRetention*time.Second - time.Second)) if got := collectAll(t, mc)["rfm_flow_bytes_total"]; got != 2000 { t.Fatalf("rfm_flow_bytes_total inside the retention = %v, want 2000", got) } c.Evict(t0.Add(collector.RollupRetention*time.Second + time.Second)) vals = collectAll(t, mc) if _, ok := vals["rfm_flow_bytes_total"]; ok { t.Fatal("idle rollup still exported after the retention window") }}
func TestCollectRollupSeriesStartAtZero(t *testing.T) { c := collector.New(time.Minute, nil, config.DefaultMaxFlows) c.SetSampleRate(10, 0) ev := collector.FlowEvent{ Ifindex: 2, Dir: 1, Proto: 6, SrcAddr: netip.MustParseAddr("192.168.1.1"), DstAddr: netip.MustParseAddr("192.168.1.2"), SrcPort: 1234, DstPort: 443, Segs: 1, Len: 100, } c.Record(ev, time.Now())
// a series whose first sample already holds its first packets loses // them to increase(), which takes the first sample for history mc := New(nil, c) want := []map[string]float64{ {"rfm_flow_packets_total": 0, "rfm_flow_bytes_total": 0, "rfm_flow_sampled_packets_total": 0, "rfm_flow_sampled_bytes_total": 0}, {"rfm_flow_packets_total": 10, "rfm_flow_bytes_total": 1000, "rfm_flow_sampled_packets_total": 1, "rfm_flow_sampled_bytes_total": 100}, } for i, w := range want { vals := collectAll(t, mc) for name, v := range w { got, ok := vals[name] if !ok || got != v { t.Fatalf("scrape %d: %s = %v (present %v), want %v", i+1, name, got, ok, v) } } }}
// asnPerSource gives every source address an asn of its own, so every flow// from a new source opens a new label tupletype asnPerSource struct{}
func (asnPerSource) Enrich(src, dst netip.Addr) (collector.Labels, collector.Labels) { b := src.As16() return collector.Labels{ASN: 64512 + uint32(b[15])}, collector.Labels{}}
// seriesID names a series by its labelsfunc seriesID(m *dto.Metric) string { var b strings.Builder for _, l := range m.GetLabel() { b.WriteString(l.GetName() + "=" + l.GetValue() + ",") } return b.String()}
func TestRollupSumsStayExactAcrossTheCap(t *testing.T) { // flows and label tuples are both capped at four c := collector.New(time.Minute, asnPerSource{}, 4) c.SetSampleRate(10, 0) mc := New(nil, c) reg := prometheus.NewRegistry() reg.MustRegister(mc)
// what a tsdb keeps, the last value of every series it scraped, a series // that leaves the scrape keeps its last value, so the sum is what // sum(increase()) adds up over all series that start at zero last := map[string]float64{} var folded float64 scrape := func() int { t.Helper() mfs, err := reg.Gather() if err != nil { t.Fatalf("gather: %v", err) } var series int for _, mf := range mfs { switch mf.GetName() { case "rfm_flow_packets_total": for _, m := range mf.GetMetric() { last[seriesID(m)] = m.GetCounter().GetValue() series++ } case "rfm_collector_folded_flows_total": folded = mf.GetMetric()[0].GetCounter().GetValue() } } return series } sum := func() float64 { var s float64 for _, v := range last { s += v } return s }
t0 := time.Now() var total float64 send := func(src byte, at time.Time) { c.Record(collector.FlowEvent{ Ifindex: 2, Dir: 0, Proto: 17, SrcPort: 5000, DstPort: 53, SrcAddr: netip.AddrFrom4([4]byte{10, 1, 0, src}), DstAddr: netip.MustParseAddr("10.0.0.1"), Segs: 1, Len: 100, }, at) total += 10 }
// four tuples fill the cap, the fifth finds no tuple a scrape has fully // shown to give up and its flow counts under empty labels for src := byte(1); src <= 5; src++ { send(src, t0) } scrape() scrape() if got := sum(); got != total { t.Fatalf("sum over all series = %v, want %v", got, total) } if folded != 1 { t.Fatalf("folded flows = %v, want 1", folded) }
// the first tuple lost its flow to the fifth and two scrapes showed all // of it, a new tuple takes its place, its series keeps what it counted // and none of it moves to the overflow series send(6, t0.Add(time.Second)) send(3, t0.Add(time.Second)) send(5, t0.Add(time.Second)) for range 2 { if n := scrape(); n > 5 { t.Fatalf("series = %d, want at most four tuples and the overflow", n) } } if got := sum(); got != total { t.Fatalf("sum over all series = %v, want %v", got, total) } if folded != 1 { t.Fatalf("folded flows = %v, want 1", folded) }}
func TestCollectIPFIXErrorsComeFromTheExporter(t *testing.T) { // the exporter refuses a record when its queue is full and counts the // loss itself, the subsystem counter shows its count and nothing else exp := &recordingExporter{err: errors.New("ipfix queue full")} c := collector.New(30*time.Second, nil, config.DefaultMaxFlows) c.SetFlowExporter(exp) t0 := time.Now() c.Record(collector.FlowEvent{ Proto: 6, SrcPort: 1, DstPort: 80, SrcAddr: netip.MustParseAddr("10.0.0.1"), DstAddr: netip.MustParseAddr("10.0.0.2"), Segs: 1, Len: 100, }, t0) c.Evict(t0.Add(time.Minute))
mc := New(nil, c) if got := errorsBySubsystem(t, mc)["ipfix"]; got != 0 { t.Fatalf("ipfix errors = %v without exporter stats, want 0", got) } mc.SetIPFIX(func() IPFIXStats { return IPFIXStats{QueueDropped: 1, SendErrors: map[string]uint64{}} }) vals := collectAll(t, mc) if got := vals["rfm_errors_total"]; got != 1 { t.Fatalf("rfm_errors_total summed over subsystems = %v, want 1 (one lost record)", got) }}
type recordingExporter struct { err error}
func (r *recordingExporter) ExportFlow(collector.ExportedFlow) error { return r.err }