flowcollector-go

Standalone Go-based flow collector for IPFIX, NetFlow, and sFlow with canonical normalization, enrichment, and multi-output delivery.

Implementation Policy

• Compatibility and correctness drive feature work.
• Normalize owns canonical field semantics.
• Behavior changes should stay explicit, incremental, and documented.

Scope Status

• flowcollector-go is the active collector implementation for the current supported project scope.
• The current codebase reflects a working collector with implemented ingest, normalize, enrich, output, packaging, and operational workflows.
• Historical planning documents may still exist in local development contexts, but the project is no longer an early-stage prototype.

Project Status

• Bootstrap service exists, starts as a long-running process, validates YAML config with strict schema checks, and boots bounded UDP receiver workers for configured listeners.
• IPFIX parser state is active in runtime (internal/ingest/ipfix): template sets, options templates, template withdrawals, and data record decoding are implemented.
• NetFlow parser state is active in runtime (internal/ingest/netflow): version-aware dispatch supports NetFlow v9 (template/options/data flowsets) and NetFlow v5 (fixed-size records).
• sFlow parser is active in runtime (internal/ingest/sflow): flow samples + expanded flow samples and raw packet header records are decoded.
• sFlow runtime compatibility decoding now includes sampled IPv4/IPv6 records, counter sample containers (2,4) with interface/ethernet counters (1,2), and extended records (1001,1002,1003,1004,1005,1006,1007,1008,1009,1010,1011,1012,1013,1014,1015,1019,1020,1021,1022,1023,1024,1025,1026,1027,1028,1029,1030,1036,1037,1040,2003,2100,2101,2102,2103,2206) mapped to compatibility flow.*/user.*/url.* fields plus conservative sflow.ext.* / sflow.counter.* metadata where no stable compatibility field exists.
• sFlow sampled IPv4/IPv6 and raw-header decoding also derives the VictoriaFlow-compatible ToS fields ip.dscp, ip.dscp.name, ip.ecn, flow.in.ip.dscp.name, and flow.in.ip.ecn.name from ip.tos.
• Parser fixture corpus exists (internal/ingest/fixtures) with snapshot regressions for IPFIX/NetFlow/sFlow decode contracts.
• Canonical event model is defined in pkg/model and reused via internal/pipeline.Event alias.
• Normalize layer exists (internal/normalize) and is responsible for canonical field mapping before output serialization.
• Normalize now enforces canonical core-field precedence and restores protected canonical values before and after enrichment stages.
• Router module exists (internal/output/router) with per-target queues, batch workers, and drop policies (drop_newest, drop_oldest).
• VictoriaLogs sink module exists (internal/output/victorialogs) with bounded queue, async batching, retry/backoff, and JSONLine posting.
• Kafka sink module exists (internal/output/kafka) with delivery retries, community-id keying, delivery telemetry counters, and optional ML prep parity (outputs.kafka.ml_prep).
• Local file sink module exists (internal/output/file) for JSONL archive output.
• Viewer sink module exists (internal/output/viewer) for debug/inspection output (ipfixcol2 viewer-like intent).
• GeoIP/ASN enrichment module exists (internal/enrich/geoasn) with optional MaxMind City/ASN DB lookups for source and destination IP metadata.
• DNS enrichment module exists (internal/enrich/dns) with reverse-lookup worker pool, TTL cache, optional custom DNS servers, and optional hosts override CSV.
• OUI vendor enrichment module exists (internal/enrich/oui) with WLAN station MAC prefix lookup (flow.wlan.sta.vendor) from external CSV.
• SNMP IF-MIB enrichment module exists (internal/enrich/snmp) with target inventory loader, async poller/cache, and interface metadata mapping (ifName/ifDescr/ifAlias/ifSpeed).
• CIDR metadata enrichment module exists (internal/enrich/cidr) with YAML loader and longest-prefix matching for flow.src.*/flow.dst.* custom fields.
• App mapping enrichment module exists (internal/enrich/app) with IPFIX applicationId (IE 95) decoding, app registry lookup, and server-port fallback mapping.
• IP reputation enrichment module exists (internal/enrich/reputation) with OSINT CSV lookup and hot-reload worker.
• Runtime wiring parses incoming packets and emits canonical events to the router/VictoriaLogs path when outputs.victorialogs.enabled=true.
• Runtime wiring loads IPFIX Information Element definitions from libfds-style XML files (ingest.ipfix.element_paths) and applies typed decoding plus ECS alias mapping (including PEN 4300/InMon enterprise fields).
• Runtime wiring supports native Kafka output when outputs.kafka.enabled=true.
• Runtime wiring supports local file output when outputs.file.enabled=true.
• Runtime wiring supports viewer/debug output when outputs.viewer.enabled=true.
• Runtime wiring supports optional GeoIP/ASN enrichment via enrichment.geoasn.
• Runtime wiring supports optional DNS enrichment via enrichment.dns.
• Runtime wiring supports optional OUI enrichment via enrichment.oui.
• Runtime wiring supports optional SNMP IF-MIB enrichment via enrichment.snmp.
• Runtime wiring supports optional CIDR metadata enrichment via enrichment.cidr.
• Runtime wiring supports optional app mapping enrichment via enrichment.app.
• Runtime wiring supports optional IP reputation enrichment via enrichment.reputation.
• Dual-output integration tests exist (internal/integration) for VictoriaLogs + Kafka isolation under sink failures and backpressure.
• Output serializers consume normalized canonical maps so VictoriaLogs, Kafka, file, and viewer share the same compatibility mapping path.
• Release workflow exists via .goreleaser.yaml and .forgejo/workflows/release.yml for multi-arch tarballs, Linux packages, checksums, and Codeberg container publication.
• Local helper scripts intentionally remain only for reputation updates and git-safety snapshots.
• Observability endpoints are available via internal/obs on observability.listen_addr:
  • /healthz (liveness)
  • /status (runtime snapshots)
  • /metrics (Prometheus text format)

Current Coverage

• IPFIX ingest, template handling, typed element decoding, normalization, enrichment, and output delivery are implemented.
• NetFlow v5 and NetFlow v9 ingest, normalization, canonical promotion, and final-output behavior are implemented for the current supported scope.
• sFlow ingest, sampled tuple decoding, extended-record decoding, normalization, enrichment, and output delivery are implemented for the current supported scope.
• Enrichment is implemented for GeoIP/ASN, DNS, OUI, SNMP IF-MIB, CIDR metadata, application mapping, and IP reputation.
• VictoriaLogs, Kafka, file, and viewer/debug outputs are implemented and share the same normalized event path.

Validation Status

• The project has been validated repeatedly using swap-based runtime comparisons and final NDJSON checks.
• Current release packaging and Codeberg release publication are operational:
  • multi-arch tar.gz
  • .deb
  • .rpm
  • checksums
  • container image publication metadata
• Remaining work is now mostly operational hardening, public-facing documentation cleanup, and future scope expansion, not basic parity catch-up for the current target scope.

Quick Start

Option A: Local Go runtime

Prerequisites:

• Go 1.24+

Run:

make run

Option B: Docker (no local Go required)

Run:

cp .env.example .env
docker compose up --build

Docker Compose mounts plugin/runtime config automatically:

• ./configs/plugins -> /data (DNS hosts, OUI, SNMP targets, CIDR metadata, app registry/ports, reputation feed)
• ./configs/libfds/system -> /etc/libfds/system and /etc/libfds/System (IPFIX system templates incl. elements/, aliases.xml, mappings.xml)
• ./configs/libfds/user/elements -> /etc/libfds/user/elements and /etc/libfds/user/Elements (IPFIX enterprise XML)
• ./artifacts/runtime -> /var/log/flowcollector (file sink output path)

If you enable GeoIP/ASN enrichment, place these files into configs/plugins/ first:

• GeoLite2-City.mmdb
• GeoLite2-ASN.mmdb

Default host port mapping (override via .env):

• NetFlow: 32055/udp -> container 2055/udp
• IPFIX: 34739/udp -> container 4739/udp
• sFlow: 36343/udp -> container 6343/udp
• Observability: 39099/tcp -> container 9099/tcp

Stop:

docker compose down

Git Safety Guardrails

Install once per clone:

make git-safety-install

Create a rescue snapshot before risky history operations (reset, rebase, cherry-pick on main):

make git-safe-snapshot

The pre-push hook blocks non-fast-forward updates to main by default.

Validate Startup

After startup, you should see logs similar to:

• {"level":"info","msg":"collector starting","service":"flowcollector-go",...}
• {"level":"info","msg":"geoasn enrichment enabled","geoip_db_path":"...","asn_db_path":"...",...} (when enabled)
• {"level":"info","msg":"oui enrichment enabled","path":"...",...} (when enabled)
• {"level":"info","msg":"snmp if-mib enrichment enabled","targets_path":"...",...} (when enabled)
• {"level":"info","msg":"cidr metadata enrichment enabled","path":"...",...} (when enabled)
• {"level":"info","msg":"app enrichment enabled","registry_path":"...","port_map_path":"...",...} (when enabled)
• {"level":"info","msg":"reputation enrichment enabled","ip_path":"...","reload_seconds":3600,...} (when enabled)
• {"level":"info","msg":"observability server started","listen_addr":":9099",...}
• {"level":"info","msg":"bootstrap complete","receivers_started":3,...}

Quick endpoint checks:

curl -fsS http://127.0.0.1:9099/healthz
curl -fsS http://127.0.0.1:9099/status | jq .
curl -fsS http://127.0.0.1:9099/metrics | head

Benchmarks

No dedicated benchmark or load-threshold helper is part of the current supported local workflow. Use direct go test -bench ... commands when benchmark work is needed.

Parser Fixtures

Run fixture snapshot regressions:

make test-fixtures

Regenerate snapshots intentionally:

make update-fixtures

Testing

Run dual-output isolation coverage (VictoriaLogs + Kafka):

make test-integration

Run docker-compose E2E signoff checks:

make e2e-signoff

Build release artifacts:

make package-release

This uses GoReleaser and writes the current tar.gz release artifacts plus checksums.txt under dist/.

For live NDJSON parity evidence review between original output and flowcollector-go, use:

python3 scripts/ndjson_diff.py --original /path/to/original.ndjson --go /path/to/flowcollector-go.ndjson

Release tags follow the governed semver scheme:

• stable releases: vX.Y.Z
• release candidates: vX.Y.Z-rcN

Date-based release tags are historical only and must not be created going forward.

The archive includes the runtime config bundle needed by the example config:

• configs/collector.example.yaml
• configs/collector.package.yaml
• configs/libfds/**
• configs/plugins/**
• deploy/systemd/**
• scripts/update_reputation.sh

Not bundled:

• external GeoLite/MaxMind databases

Release

Releases are CI-owned and idempotent (see docs/release/RELEASE_WORKFLOW.md). The Forgejo/Codeberg release body is published from the matching CHANGELOG.md section for the tagged version. Stable release tags use vX.Y.Z; release candidates use vX.Y.Z-rcN only when explicitly justified; date-based tags are historical only and must not be created going forward.

Deferred Renovate updates

• PR #4 fix(deps): update module github.com/oschwald/geoip2-golang to v2 reviewed on 2026-04-23
• Status: intentionally deferred
• Reason: major runtime dependency update; real source migration from v1 imports is still missing; the stale branch must not be merged as-is because it would also revert current governed release hardening
• Next action: replay only after explicit code migration work from v1 to v2 on current main

The tar.gz bundle now also carries the packaged systemd material so manual installs can mirror the Linux package layout:

• deploy/systemd/flowcollector.service
• deploy/systemd/flowcollector-reputation-update.service
• deploy/systemd/flowcollector-reputation-update.timer
• deploy/systemd/flowcollector-reputation-update.env.example

Package builds (.deb / .rpm) install:

• binary: /usr/bin/flowcollector
• package config: /etc/flowcollector/collector.yaml
• libfds definitions: /etc/flowcollector/libfds/...
• plugin/sample data: /etc/flowcollector/plugins/...
• main service unit: /usr/lib/systemd/system/flowcollector.service
• reputation update service/timer:
  • /usr/lib/systemd/system/flowcollector-reputation-update.service
  • /usr/lib/systemd/system/flowcollector-reputation-update.timer
  • /etc/default/flowcollector-reputation

GeoIP/ASN Enrichment

Enable enrichment.geoasn.enabled=true in configs/collector.example.yaml and provide at least one DB path:

• enrichment.geoasn.geoip_db_path for city/country/location fields
• enrichment.geoasn.asn_db_path for ASN number/org fields

When active, source/destination fields like source.geo.country_iso_code, source.as.number, destination.geo.*, and destination.as.* are added when lookups succeed. For private IPv4 sides, parity compatibility emits ASN 0 plus PRIVATE org/label without a database lookup. The compatibility field surface also includes directional ASN labels (flow.src.as.label, flow.dst.as.label) plus top-level as.label when label data is present.

DNS Enrichment

Enable enrichment.dns.enabled=true in configs/collector.example.yaml.

Optional tuning:

• enrichment.dns.servers (CSV of DNS server IPs, optional :port, empty = system resolver)
• enrichment.dns.hosts_path (CSV ip,hostname override)
• enrichment.dns.timeout_ms
• enrichment.dns.cache_ttl_seconds
• enrichment.dns.cache_max_entries
• enrichment.dns.workers
• enrichment.dns.queue_size

When active, PTR names are added as source.dns.ptr_name and destination.dns.ptr_name when lookup/override succeeds.

OUI Vendor Enrichment

Enable enrichment.oui.enabled=true and provide:

• enrichment.oui.path (CSV in OUI,Vendor format, for example AA:BB:CC,Vendor Name)

When active, flow.wlan.sta.vendor is added when flow.wlan.sta.mac is present and the OUI is known. Locally administered/randomized MACs are skipped.

SNMP IF-MIB Enrichment

Enable enrichment.snmp.enabled=true and provide:

• enrichment.snmp.targets_path (CSV format: exporter_ip,snmp_host,community,version,port)

Optional tuning:

• enrichment.snmp.poll_interval_seconds
• enrichment.snmp.cache_ttl_seconds
• enrichment.snmp.timeout_ms
• enrichment.snmp.retries
• enrichment.snmp.workers

Behavior:

• Polls IF-MIB columns ifName, ifDescr, ifAlias, ifSpeed, and ifHighSpeed.
• Maps interface metadata by exporter_ip + ifIndex to:
  • flow.in.netif.{name,descr,alias,speed}
  • flow.out.netif.{name,descr,alias,speed}
• Falls back to flow.in/out.netif.name = "index: <ifIndex>" when only interface index is known.

CIDR Metadata Enrichment

Enable enrichment.cidr.enabled=true and provide:

• enrichment.cidr.path (YAML format, see configs/plugins/ip-metadata.yaml)

Behavior:

• Applies longest-prefix CIDR matching for source/destination IPs.
• Adds compatibility fields under flow.src.* and flow.dst.*:
  • *.subnet.name
  • *.sec.zone
  • *.tags
  • *.geo.* (country.code/name, city.name, region.name, continent.code, tz.name, loc.coord)
• Also writes ECS-style geo keys (source.geo.* / destination.geo.*) so CIDR-provided geo can override GeoIP lookups in the enrichment chain.

App Mapping Enrichment

Enable enrichment.app.enabled=true in configs/collector.example.yaml.

Optional inputs:

• enrichment.app.registry_path (CSV with columns app_id,name,category)
• enrichment.app.port_map_path (CSV with columns port,protocol,name)

Behavior:

• Decodes ipfix.ie.95 into app.id (engine..selector), app.class.name, and app.source.
• Resolves app.name and optional app.category via registry CSV when present.
• Adds flow.app.name from server-port mapping (built-in map; optional CSV overrides).

IP Reputation Enrichment

Enable enrichment.reputation.enabled=true and set enrichment.reputation.ip_path to a CSV file using:

ip,category,score,providers,malware,tags,reference

Optional tuning:

• enrichment.reputation.reload_seconds (0 disables reload checks)

When active, the pipeline emits threat.* fields (for example threat.enrich.status, threat.indicator.*, threat.feed.name) with destination-IP-first matching parity (flow.dst.ip.addr, then flow.src.ip.addr).

Build/update reputation_ip.csv from public OSINT feeds:

make update-reputation

Optional custom output directory:

make update-reputation OUTPUT_DIR=/tmp/reputation

Run cyclic via cron (every 6h):

0 */6 * * * cd /path/to/flowcollector-go && ./scripts/update_reputation.sh ./configs/plugins >> /var/log/flowcollector-reputation.log 2>&1

Run cyclic via systemd timer:

sudo systemctl daemon-reload
sudo systemctl enable --now flowcollector-reputation-update.timer

Run the packaged collector via systemd:

sudo systemctl daemon-reload
sudo systemctl enable --now flowcollector.service

Volume reachability check (configs/plugins is mounted as /data):

docker compose exec flowcollector ls -l /data/reputation_ip.csv
docker compose exec flowcollector head -n 3 /data/reputation_ip.csv

If enrichment.reputation.reload_seconds > 0, updated files are picked up without collector restart.

IPFIX Element Registry (Enterprise Templates)

IPFIX field decoding can load libfds element XML definitions and map values with correct data types.

Config path:

• ingest.ipfix.element_paths (YAML list of directories and/or XML files)

Default search paths used when config list is empty:

• /etc/libfds/system/elements
• /etc/libfds/System
• /etc/libfds/user/elements
• /etc/libfds/user/Elements

The bundled system registry includes IANA/core libfds definitions plus vendor XMLs for:

• juniper
• ixia
• f5
• sonicwall
• barracuda
• ntop
• a10

With this registry, enterprise IEs (notably PEN 4300, plus XML-backed vendor PENs when present in traffic) are decoded and mapped to ECS-compatible fields such as:

• flow.in.vlan.tag.id, flow.out.vlan.tag.id
• flow.nat.*
• flow.mpls.*
• flow.vlan.tunnel.stack
• flow.vxlan.vni.*

Numeric raw keys (ipfix.ie.*) remain available.

sFlow Extended Compatibility

The runtime now decodes sFlow sampled tuple records, approved counter sample containers, and extended records from parsed sFlow datagrams:

• sampled tuple records: 3 (sampled_ipv4), 4 (sampled_ipv6)
• counter sample containers: 2 (counter_sample), 4 (expanded_counter_sample)
• counter records: 1 (generic interface counters), 2 (ethernet interface counters)
• extended records: 1001,1002,1003,1004,1005,1006,1007,1008,1009,1010,1011,1012,1013,1014,1015,1019,1020,1021,1022,1023,1024,1025,1026,1027,1028,1029,1030,1036,1037,1040,2003,2100,2101,2102,2103,2206

This provides compatibility fields for VLAN, NAT, MPLS, BGP, URL/user metadata, VXLAN VNI data, WLAN metadata, ACL IDs, queue metadata, conservative tunnel/decapsulation/socket/proxy/HTTP metadata, and sflow.counter.* fields for interface and ethernet counters.

VictoriaLogs Stream Fields

Configure stream field parity with the original implementation via:

• outputs.victorialogs.stream_fields
• outputs.victorialogs.time_field

Supported stream_fields syntax:

• DIRECT: flow.app.name
• ALIAS: host={flow.export.host.name}
• STATIC: job=integrations/netflow-receiver

Static values also support inline placeholders (example: stream=netflow-{flow.id}). The sink automatically appends _stream_fields and _time_field query parameters to the VictoriaLogs request URL. VictoriaLogs/Kafka/file/viewer JSON now also include ECS-oriented exporter metadata defaults (flow.export.*) and NetFlow version fields (netflow.version, flow.export.version.ver) when applicable.

Compatibility Fields

The output builder derives compatibility fields when source/destination tuple data is available, including:

• flow.locality (private / public / mixed)
• flow.isServer (source / destination)
• flow.server.* and flow.client.* mirror fields
• flow.src.* / flow.dst.* compatibility keys derived from ECS tuple fields
• plus direction/protocol/sampling compatibility fields where source data exists: flow.direction*, flow.ip.protocol, flow.meter.sample.*, flow.in.*, and flow.collect.timestamp

Kafka ML Prep

When outputs.kafka.ml_prep=true, Kafka JSON output adds:

• ml_key: <exporter>|<proto>|<server_port>
• ml: true

Kafka key mode is configurable via outputs.kafka.key_mode:

• community_id (default): Kafka message key is flow.community.id when present.
• ml_key: Kafka message key uses the same ML key heuristic (<exporter>|<proto>|<server_port>).

Current downstream ML examples use the neutral Kafka topic normalized.ml.events. That topic is shared with other producers and consumed centrally by flowcollector-ml; it is no longer a flowcollector-go-only path. flow.* compatibility fields remain the scorer-facing canonical ML contract on that topic.

Main Paths

• Entry point: cmd/collector/main.go
• IPFIX element registry + ECS alias mapping: cmd/collector/ipfix_elements.go
• sFlow extended compatibility decoder: cmd/collector/sflow_extended.go
• Config loader: internal/config/config.go
• IPFIX parser: internal/ingest/ipfix/parser.go
• NetFlow v9 parser: internal/ingest/netflow/parser_v9.go
• NetFlow v5 parser: internal/ingest/netflow/parser_v5.go
• sFlow parser: internal/ingest/sflow/parser.go
• Canonical model: pkg/model/model.go
• OUI enrichment: internal/enrich/oui/oui.go
• SNMP IF-MIB enrichment: internal/enrich/snmp/snmp.go
• CIDR metadata enrichment: internal/enrich/cidr/cidr.go
• App mapping enrichment: internal/enrich/app/app.go
• GeoIP/ASN enrichment: internal/enrich/geoasn/geoasn.go
• DNS enrichment: internal/enrich/dns/dns.go
• Reputation enrichment: internal/enrich/reputation/reputation.go
• Output router: internal/output/router/router.go
• VictoriaLogs sink: internal/output/victorialogs/sink.go
• Kafka sink: internal/output/kafka/sink.go
• File sink: internal/output/file/sink.go
• Viewer sink: internal/output/viewer/sink.go
• Example config: configs/collector.example.yaml
• CIDR metadata example: configs/plugins/ip-metadata.yaml