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 output includes sampled IPv4/IPv6 flow records and supported 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). Counter sample containers (2,4) and interface/ethernet counters (1,2) are parsed and have a normalizer API, but are not dispatched by the collector runtime yet.
• 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 authoritative hosts.csv support.
• 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

DNS enrichment is enabled by default and can be disabled with enrichment.dns.enabled=false.

Optional tuning:

• enrichment.dns.servers (CSV of DNS server IPs, optional :port, empty = system resolver)
• enrichment.dns.hosts_path (CSV ip,hostname hostname source)
• enrichment.dns.timeout_ms
• enrichment.dns.cache_ttl_seconds
• enrichment.dns.negative_cache_ttl_seconds
• enrichment.dns.cache_max_entries
• enrichment.dns.workers
• enrichment.dns.queue_size
• enrichment.dns.skip_private_ptr (when true, unknown private addresses do not generate PTR lookups)

Recommended production model: generate a canonical hosts.csv from the environment inventory, configure it with enrichment.dns.hosts_path, and set enrichment.dns.skip_private_ptr=true for private networks. The file is checked before both DNS caches, so it is authoritative for known private hosts and also overrides any stale cached PTR result. Existing successful DNS enrichment is preserved for public addresses and any private addresses that are not suppressed.

Resolution order is:

1. hosts.csv
2. positive DNS cache
3. negative DNS cache
4. stop when skip_private_ptr=true and the IP is private
5. PTR lookup for remaining addresses

When active, names are added as source.dns.ptr_name and destination.dns.ptr_name when lookup or hosts.csv resolution succeeds. Failed PTR lookups are cached for the negative-cache TTL so repeated NXDOMAIN, NODATA, timeout, SERVFAIL, and resolver errors for the same IP do not repeatedly hit DNS.

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:

• Core normalization always preserves flow.in/out.netif.index and adds flow.in/out.netif.name = "index: <ifIndex>" when a non-zero numeric index is present but no real interface name was exported.
• 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}
• Replaces only the matching index: <ifIndex> fallback with an exporter-scoped real name; explicit names carried by the flow record remain authoritative.

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
  • *.site.* (country.code/name, city.name, region.name, continent.code, tz.name, loc.coord)
• CIDR site/location metadata is intentionally project-specific under flow.src.site.* / flow.dst.site.*. This keeps local inventory/site ownership separate from ECS GeoIP fields.
• CIDR enrichment does not write source.geo.*, destination.geo.*, flow.src.geo.*, or flow.dst.geo.*. Those namespaces remain reserved for GeoIP/MaxMind enrichment and ElastiFlow GeoIP compatibility mirrors.
• Metadata files should use site.* keys. Existing geo.* keys in CIDR metadata files are accepted as input aliases and are written to flow.*.site.*, not ECS Geo fields.

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; optional category)

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).
• Uses the optional port-map category column as a low-confidence fallback for app.category only when app.category is empty. Protocol-specific port rows win over protocol-agnostic rows, and existing categories from IPFIX/NBAR/registry enrichment are preserved.

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 are decoded only under their declared PEN. Exact PEN 4300 aliases provide compatibility fields such as:

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

An enterprise IE never inherits an IANA alias merely because its numeric ID is the same. XML-backed enterprise fields use their registry fallback name, and unknown fields use a PEN-qualified fallback.

sFlow Extended Compatibility

The parser and normalizer decode sFlow sampled tuple records, 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.

The runtime currently dispatches flow samples only. It does not emit parsed counter samples. Raw sampled headers are decoded according to their declared sFlow header protocol (Ethernet, IPv4, or IPv6), and exporter identity uses the datagram agent_address plus sub-agent context rather than the UDP relay address.

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

NetFlow v9 Options scope fields

NetFlow v9 Options Template scope fields are not Information Elements. The parser retains their scope marker through Options Data Record decoding, and normalization emits their raw values under netflow.v9.scope.<scope-type> (or netflow.v9.scope.enterprise.<enterprise-id>.<scope-type> when applicable). They never enter the ordinary netflow.v9.field.<ie-id> promotion path, so a scope type cannot create traffic counters or other IE-derived fields.

Non-scope option fields retain their raw IE representation but are marked as a metadata record and cannot enter tuple, counter, sampling, or timestamp compatibility normalization. For example, application name IE 96 remains available as netflow.v9.field.96; the existing IE 95 application-ID enrichment input remains available as ipfix.ie.95.

IPFIX Options and enterprise fields

IPFIX Options scope fields are emitted under ipfix.scope.<ie-id> (or ipfix.scope.enterprise.<pen>.<ie-id>). Non-scope option fields remain under their ipfix.ie.* raw key. The whole Options Data Record is metadata-only, so neither scope nor option values can create flow bytes, packets, addresses, ports, sampling correction, or timestamps.

IPFIX and NetFlow v9 template caches include the exporter UDP source port in addition to the observation/source domain and template ID. This prevents a template from one transport context from decoding another exporter's data.

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