From d36eb94b5ab01b30df3803570bb9885123228bf5 Mon Sep 17 00:00:00 2001 From: Dylan Myers Date: Thu, 24 Sep 2026 16:21:35 -0400 Subject: [PATCH] feat(flow): NetFlow/sFlow/IPFIX generators + FlowRecord signal (PIPE-977, BP-474) Assisted-By: Claude Opus 4.8 --- cmd/blitz/main.go | 13 + docker/README.md | 1 + docker/docker-compose.telemetry-generator.yml | 15 ++ docs/generator/flow.md | 142 +++++++++++ embed/consumer.go | 8 + embed/host.go | 3 + embed/otelpdata/flow.go | 108 ++++++++ embed/otelpdata/flow_test.go | 51 ++++ embed/record.go | 59 ++++- embed/runner.go | 3 +- flow/ipfix/ipfix.go | 240 ++++++++++++++++++ flow/ipfix/ipfix_test.go | 50 ++++ flow/ipfix/roundtrip_test.go | 85 +++++++ flow/netflow/roundtrip_test.go | 97 +++++++ flow/netflow/v5.go | 114 +++++++++ flow/netflow/v5_test.go | 94 +++++++ flow/netflow/v9.go | 232 +++++++++++++++++ flow/netflow/v9_test.go | 53 ++++ flow/sflow/sflow.go | 96 +++++++ flow/sflow/sflow_test.go | 60 +++++ flow/wire/wire.go | 32 +++ generator/flow/flow.go | 154 +++++++++++ generator/flow/scenario.go | 103 ++++++++ generator/flow/scenario_test.go | 36 +++ go.mod | 1 + go.sum | 2 + internal/config/generator.go | 11 +- internal/config/generator_flow.go | 47 ++++ internal/config/output.go | 10 +- internal/config/output_flow.go | 92 +++++++ internal/config/output_flow_test.go | 38 +++ internal/config/override.go | 9 + internal/config/override_test.go | 55 ++++ internal/dispatch/embed.go | 14 + output/flow/flow.go | 180 +++++++++++++ output/flow/flow_test.go | 45 ++++ output/otlp_grpc/flows.go | 31 +++ package/completions/blitz.bash | 180 +++++++++++++ telemetry/telemetry.go | 4 +- 39 files changed, 2563 insertions(+), 5 deletions(-) create mode 100644 docs/generator/flow.md create mode 100644 embed/otelpdata/flow.go create mode 100644 embed/otelpdata/flow_test.go create mode 100644 flow/ipfix/ipfix.go create mode 100644 flow/ipfix/ipfix_test.go create mode 100644 flow/ipfix/roundtrip_test.go create mode 100644 flow/netflow/roundtrip_test.go create mode 100644 flow/netflow/v5.go create mode 100644 flow/netflow/v5_test.go create mode 100644 flow/netflow/v9.go create mode 100644 flow/netflow/v9_test.go create mode 100644 flow/sflow/sflow.go create mode 100644 flow/sflow/sflow_test.go create mode 100644 flow/wire/wire.go create mode 100644 generator/flow/flow.go create mode 100644 generator/flow/scenario.go create mode 100644 generator/flow/scenario_test.go create mode 100644 internal/config/generator_flow.go create mode 100644 internal/config/output_flow.go create mode 100644 internal/config/output_flow_test.go create mode 100644 output/flow/flow.go create mode 100644 output/flow/flow_test.go create mode 100644 output/otlp_grpc/flows.go diff --git a/cmd/blitz/main.go b/cmd/blitz/main.go index edf95ecc..8bce201d 100644 --- a/cmd/blitz/main.go +++ b/cmd/blitz/main.go @@ -32,6 +32,7 @@ import ( "github.com/observiq/blitz/internal/telemetry/traces" "github.com/observiq/blitz/output" fileout "github.com/observiq/blitz/output/file" + flowout "github.com/observiq/blitz/output/flow" hecout "github.com/observiq/blitz/output/hec" "github.com/observiq/blitz/output/nop" otlpgrpc "github.com/observiq/blitz/output/otlp_grpc" @@ -367,6 +368,13 @@ func run(cmd *cobra.Command, args []string) error { logger.Error("Failed to create HEC output", zap.Error(err)) return err } + case config.OutputTypeFlow: + f := cfg.Output.Flow + outputInstance, err = flowout.New(logger, f.Host, strconv.Itoa(f.Port), flowout.Protocol(f.Protocol), f.Vendor, f.AgentIP, tel) + if err != nil { + logger.Error("Failed to create flow output", zap.Error(err)) + return err + } default: logger.Error("Invalid output type", zap.String("type", string(cfg.Output.Type))) return fmt.Errorf("invalid output type: %s", cfg.Output.Type) @@ -501,6 +509,11 @@ func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Out if tw, ok := out.(output.TraceWriter); ok { consumers.TraceConsumer = output.WriterAsTraceConsumer(tw, tel) } + // The flow output implements FlowConsumer directly (it UDP-encodes flows), + // so wire it through when the configured output supports flows. + if fc, ok := out.(embed.FlowConsumer); ok { + consumers.FlowConsumer = fc + } // Pass the embedded library so an embed_library build resolves package // sources; without the tag FS() is empty and resolution uses disk (PIPE-1445). mod, err := dispatch.ForEmbed(logger, genCfg, consumers, embeddedlibrary.FS(), env, tel) diff --git a/docker/README.md b/docker/README.md index 141dabd0..9d60c9d4 100644 --- a/docker/README.md +++ b/docker/README.md @@ -109,6 +109,7 @@ docker compose -f docker/docker-compose.telemetry-generator.yml up -d | `blitz-hostmetrics-windows` | Host Metrics | Synthetic host metrics (CPU, memory, disk, etc.) for Windows | | `blitz-traces` | Traces | Synthetic distributed traces (HTTP + DB spans) | | `blitz-fix` | FIX | FIX protocol messages (4.2 / 4.4 / 5.0 SP2) across 10 asset categories — NewOrderSingle, ExecutionReport, cancel/replace/status | +| `blitz-flow` | Network Flow | Synthetic NetFlow/IPFIX/sFlow records exported over UDP | ## Running Individual Generators diff --git a/docker/docker-compose.telemetry-generator.yml b/docker/docker-compose.telemetry-generator.yml index 7780a7e2..dfd58e43 100644 --- a/docker/docker-compose.telemetry-generator.yml +++ b/docker/docker-compose.telemetry-generator.yml @@ -230,6 +230,21 @@ services: BLITZ_OUTPUT_OTLPGRPC_HOST: bdot-collector BLITZ_OUTPUT_OTLPGRPC_PORT: "4317" + # Network flow records (NetFlow/IPFIX/sFlow) exported over UDP. Point it at a + # collector with a netflow/sflow receiver (e.g. bdot-collector with a + # netflowreceiver, or a standalone nfcapd/sflowtool). + blitz-flow: + <<: *blitz-common + environment: + BLITZ_GENERATOR_TYPE: flow + BLITZ_GENERATOR_FLOW_WORKERS: ${BLITZ_WORKERS:-1} + BLITZ_GENERATOR_FLOW_RATE: ${BLITZ_RATE:-1s} + BLITZ_GENERATOR_FLOW_SCENARIO: ${FLOW_SCENARIO:-default} + BLITZ_OUTPUT_TYPE: flow + BLITZ_OUTPUT_FLOW_HOST: ${FLOW_COLLECTOR_HOST:-bdot-collector} + BLITZ_OUTPUT_FLOW_PORT: ${FLOW_COLLECTOR_PORT:-2055} + BLITZ_OUTPUT_FLOW_PROTOCOL: ${FLOW_PROTOCOL:-netflow-v9} + networks: telemetry-net: driver: bridge diff --git a/docs/generator/flow.md b/docs/generator/flow.md new file mode 100644 index 00000000..f2fa8a05 --- /dev/null +++ b/docs/generator/flow.md @@ -0,0 +1,142 @@ +# Network Flow Generator + +The flow generator produces network flow records — the classic 5-tuple (source +and destination IP/port plus protocol) with byte/packet counters and routing +dimensions — and the flow output exports them over UDP in the wire format a +collector expects. FlowRecord is blitz's fourth signal type, alongside logs, +metrics, and traces. + +The generator is protocol-agnostic: it yields FlowRecords, and the flow output +chooses the wire format. So one generator drives every exporter format. + +## Protocol coverage + +| Wire format | `protocol` value | Notes | +|-------------|------------------|-------| +| NetFlow v5 | `netflow-v5` | Fixed 48-byte records, IPv4 only, ≤30 flows/packet | +| NetFlow v9 | `netflow-v9` | Template flowsets, periodic template resend | +| IPFIX | `ipfix` | RFC 7011; 64-bit counters, absolute ms timestamps | +| sFlow v5 | `sflow` | Flow samples carrying sampled-IPv4 records | + +NetFlow v1/v7/v8 are intentionally omitted (superseded; negligible residual +deployment). + +## Vendor flavors + +A vendor flavor makes the export genuinely vendor-distinct on the wire — not a +metadata tag. It selects a distinct encoding path whose distinguishing element a +decoder (e.g. netsampler/goflow2, which the collector `netflowreceiver` embeds) +reads back. The flavor is a single field, so flavors are mutually exclusive by +construction, and each must be paired with the protocol whose format carries its +signature. + +How the distinction is carried depends on the format: + +- **IPFIX** is the only format with a decoder-readable enterprise element: a + template field with the enterprise bit (`0x8000`) set, followed by the + vendor's 4-byte IANA Private Enterprise Number (PEN). goflow2 surfaces this as + the decoded field's PEN. AppFlow, jFlow, and cflowd use this. +- **NetFlow v9** has no PEN mechanism, so a v9 flavor is distinguished by a + proprietary field type id in the template (goflow2 decodes the field by its + type). NetStream and rFlow use this. + +A vendor may be paired with any format it supports in the real world; +`Validate()` rejects an unsupported pairing. Because NetFlow v5 is a fixed +layout with no vendor mechanism, a v5 export is byte-identical across vendors +(the pairing is allowed for realism but carries no vendor-distinct element). + +| `vendor` | Vendor | Supported formats | IPFIX PEN | v9 field type | +|-------------|--------|-------------------|-----------|---------------| +| `jflow` | Juniper | `netflow-v5`, `netflow-v9`, `ipfix` | 2636 | `0x9003` | +| `netstream` | Huawei | `netflow-v5`, `netflow-v9`, `ipfix` | 2011 | `0x9001` | +| `cflowd` | Nokia/Alcatel-Lucent | `netflow-v5`, `netflow-v9`, `ipfix` | 6527 | `0x9004` | +| `appflow` | Citrix | `ipfix` | 5951 | — | +| `rflow` | Redback/Ericsson | `netflow-v5`, `netflow-v9` | — | `0x9002` | + +Format support sources: Juniper Flow Monitoring (jFlow v5/v9/IPFIX), Huawei +NetStream configuration guide (v5/v9/IPFIX), Nokia SR OS cflowd (v5/v8/v9/IPFIX; +v8 aggregation-only, excluded), Citrix AppFlow (IPFIX application), Redback/Ericsson +SmartEdge rFlow (v5/v9). Round-trip verified against netsampler/goflow2 v2.2.6, +the decoder the collector `netflowreceiver` embeds. + +## OTLP output + +Beyond the UDP wire formats, flows can be emitted as OTLP logs: set +`output.type: otlp-grpc` with `generator.type: flow`. Each FlowRecord is +projected to an OTLP `LogRecord` with the same network-flow semantic-convention +attributes (`source.address`, `destination.port`, `network.transport`, +`flow.io.bytes`, …) the collector's `netflowreceiver` emits for a decoded flow, +so a downstream consumer sees the same log a real collector would produce. This +path uses raw OTLP protobuf and adds no `collector/pdata` dependency. + +## Transport + +Flow export is UDP only — every flow collector listens on UDP. The flow output +dials `host:port` and sends each encoded packet. + +## Scenario presets + +The `scenario` setting shapes the byte/packet distribution and source/destination +spread: + +| `scenario` | Shape | +|---------------|-------| +| `default` | Broad mix of flow sizes | +| `wan-edge` | Few, large, bidirectional flows | +| `datacenter` | Many small flows | +| `ddos-target` | Many sources onto a single destination | + +## Configuration + +Generator: + +| YAML Path | Flag | Env | Default | Description | +|----------------------------|-------------------------------|---------------------------------|-----------|-------------| +| `generator.flow.workers` | `--generator-flow-workers` | `BLITZ_GENERATOR_FLOW_WORKERS` | `1` | Worker goroutines | +| `generator.flow.rate` | `--generator-flow-rate` | `BLITZ_GENERATOR_FLOW_RATE` | `1s` | Interval per worker | +| `generator.flow.scenario` | `--generator-flow-scenario` | `BLITZ_GENERATOR_FLOW_SCENARIO` | `default` | Traffic-shape preset | +| `generator.flow.seed` | `--generator-flow-seed` | `BLITZ_GENERATOR_FLOW_SEED` | `-1` | RNG seed (negative randomizes) | + +Output: + +| YAML Path | Flag | Env | Default | Description | +|------------------------|-----------------------------|-------------------------------|--------------|-------------| +| `output.flow.host` | `--output-flow-host` | `BLITZ_OUTPUT_FLOW_HOST` | `127.0.0.1` | Collector host | +| `output.flow.port` | `--output-flow-port` | `BLITZ_OUTPUT_FLOW_PORT` | `2055` | Collector UDP port | +| `output.flow.protocol` | `--output-flow-protocol` | `BLITZ_OUTPUT_FLOW_PROTOCOL` | `netflow-v9` | Wire format | +| `output.flow.vendor` | `--output-flow-vendor` | `BLITZ_OUTPUT_FLOW_VENDOR` | `""` | Optional vendor flavor | +| `output.flow.agentIP` | `--output-flow-agentip` | `BLITZ_OUTPUT_FLOW_AGENTIP` | `""` | sFlow exporter agent IP | + +## Example + +```yaml +generator: + type: flow + flow: + workers: 2 + rate: 500ms + scenario: datacenter +output: + type: flow + flow: + host: collector.example.com + port: 2055 + protocol: netflow-v9 +``` + +## Embedding + +A host consumes flows in-process by implementing `embed.FlowConsumer`: + +```go +type myConsumer struct{} + +func (myConsumer) ConsumeFlows(ctx context.Context, recs []embed.FlowRecord) error { + for _, r := range recs { + // r.SrcIP, r.DstIP, r.SrcPort, r.DstPort, r.Protocol, r.Bytes, r.Packets ... + } + return nil +} + +host := embed.Host{Flows: myConsumer{}} +``` diff --git a/embed/consumer.go b/embed/consumer.go index 0f855195..8488693d 100644 --- a/embed/consumer.go +++ b/embed/consumer.go @@ -23,3 +23,11 @@ type MetricConsumer interface { type TraceConsumer interface { ConsumeTraces(ctx context.Context, spans []Span) error } + +// FlowConsumer consumes batches of flow records produced by blitz flow +// modules (netflow, ipfix, sflow). +// +// Implementations must be safe for concurrent calls. +type FlowConsumer interface { + ConsumeFlows(ctx context.Context, records []FlowRecord) error +} diff --git a/embed/host.go b/embed/host.go index 9aab3264..56bce8b9 100644 --- a/embed/host.go +++ b/embed/host.go @@ -20,6 +20,9 @@ type Host struct { // Traces is the destination for spans. Nil means spans are dropped. Traces TraceConsumer + // Flows is the destination for flow records. Nil means flows are dropped. + Flows FlowConsumer + // Resource is the per-session base resource attributes blitz applies // to every emitted record before module-level overrides merge on top. // diff --git a/embed/otelpdata/flow.go b/embed/otelpdata/flow.go new file mode 100644 index 00000000..b14b84d1 --- /dev/null +++ b/embed/otelpdata/flow.go @@ -0,0 +1,108 @@ +// Package otelpdata converts blitz signals to raw OpenTelemetry protobuf +// (go.opentelemetry.io/proto/otlp/*). It deliberately does NOT depend on +// go.opentelemetry.io/collector/pdata — blitz's OTLP path is built on the raw +// proto types, which are already a dependency. +// +// FlowToLogRecord projects a FlowRecord onto an OTLP LogRecord using the same +// attribute keys the collector's netflowreceiver emits for a decoded flow +// (receiver/netflowreceiver/parser.go addMessageAttributes), so a downstream +// consumer sees the same log a real collector would produce from the equivalent +// wire packet. +package otelpdata + +import ( + "net" + + "github.com/observiq/blitz/embed" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + logspb "go.opentelemetry.io/proto/otlp/logs/v1" +) + +// OTel semantic-convention attribute keys shared with netflowreceiver. +const ( + attrSourceAddress = "source.address" + attrSourcePort = "source.port" + attrDestinationAddress = "destination.address" + attrDestinationPort = "destination.port" + attrNetworkTransport = "network.transport" + attrNetworkType = "network.type" +) + +// FlowToLogRecord converts a FlowRecord to an OTLP LogRecord with network-flow +// semantic-convention attributes matching netflowreceiver's output. +func FlowToLogRecord(f embed.FlowRecord) *logspb.LogRecord { + attrs := []*commonpb.KeyValue{ + str(attrSourceAddress, ipString(f.SrcIP)), + i64(attrSourcePort, int64(f.SrcPort)), + str(attrDestinationAddress, ipString(f.DstIP)), + i64(attrDestinationPort, int64(f.DstPort)), + str(attrNetworkTransport, transportName(f.Protocol)), + str(attrNetworkType, "ipv4"), + i64("flow.io.bytes", counter(f.Bytes)), + i64("flow.io.packets", counter(f.Packets)), + i64("flow.tcp_flags", int64(f.TCPFlags)), + i64("flow.ip_tos", int64(f.TOS)), + i64("flow.in_if", int64(f.InputIface)), + i64("flow.out_if", int64(f.OutputIface)), + i64("flow.src_as", int64(f.SrcAS)), + i64("flow.dst_as", int64(f.DstAS)), + i64("flow.sampling_rate", int64(f.SamplingRate)), + } + if len(f.NextHop) > 0 { + attrs = append(attrs, str("flow.next_hop", ipString(f.NextHop))) + } + + rec := &logspb.LogRecord{ + SeverityNumber: logspb.SeverityNumber_SEVERITY_NUMBER_INFO, + SeverityText: "INFO", + Attributes: attrs, + } + if !f.StartTime.IsZero() { + rec.TimeUnixNano = uint64(f.StartTime.UnixNano()) // #nosec G115 -- wall-clock ns fits uint64 for any realistic time + } + if !f.EndTime.IsZero() { + rec.Attributes = append(rec.Attributes, i64("flow.end", f.EndTime.UnixNano())) + } + if !f.StartTime.IsZero() { + rec.Attributes = append(rec.Attributes, i64("flow.start", f.StartTime.UnixNano())) + } + return rec +} + +func str(k, v string) *commonpb.KeyValue { + return &commonpb.KeyValue{Key: k, Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: v}}} +} + +func i64(k string, v int64) *commonpb.KeyValue { + return &commonpb.KeyValue{Key: k, Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_IntValue{IntValue: v}}} +} + +// counter narrows a uint64 flow counter to the int64 an OTLP attribute holds, +// matching netflowreceiver, which does the same int64(pm.Bytes) conversion. +func counter(v uint64) int64 { + return int64(v) // #nosec G115 -- semconv counters are int64; matches netflowreceiver +} + +func ipString(ip net.IP) string { + if len(ip) == 0 { + return "" + } + return ip.String() +} + +// transportName maps an IP protocol number to the OTel network.transport value, +// matching netflowreceiver's getTransportName for the common cases. +func transportName(proto uint8) string { + switch proto { + case 1: + return "icmp" + case 6: + return "tcp" + case 17: + return "udp" + case 58: + return "ipv6-icmp" + default: + return "unknown" + } +} diff --git a/embed/otelpdata/flow_test.go b/embed/otelpdata/flow_test.go new file mode 100644 index 00000000..0ca3f344 --- /dev/null +++ b/embed/otelpdata/flow_test.go @@ -0,0 +1,51 @@ +package otelpdata + +import ( + "net" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" +) + +func attrs(rec interface{ GetAttributes() []*commonpb.KeyValue }) map[string]*commonpb.AnyValue { + m := map[string]*commonpb.AnyValue{} + for _, kv := range rec.GetAttributes() { + m[kv.Key] = kv.Value + } + return m +} + +func TestFlowToLogRecordSemconv(t *testing.T) { + f := embed.FlowRecord{ + SrcIP: net.IPv4(10, 0, 0, 1), DstIP: net.IPv4(8, 8, 8, 8), + SrcPort: 54321, DstPort: 443, Protocol: 6, TCPFlags: 0x12, + Bytes: 4200, Packets: 42, InputIface: 2, OutputIface: 3, + SrcAS: 64500, DstAS: 15133, TOS: 0, SamplingRate: 1024, + NextHop: net.IPv4(10, 0, 0, 254), + StartTime: time.Unix(1700000000, 0), EndTime: time.Unix(1700000005, 0), + } + rec := FlowToLogRecord(f) + a := attrs(rec) + + require.Equal(t, "10.0.0.1", a[attrSourceAddress].GetStringValue()) + require.Equal(t, int64(54321), a[attrSourcePort].GetIntValue()) + require.Equal(t, "8.8.8.8", a[attrDestinationAddress].GetStringValue()) + require.Equal(t, int64(443), a[attrDestinationPort].GetIntValue()) + require.Equal(t, "tcp", a[attrNetworkTransport].GetStringValue()) + require.Equal(t, "ipv4", a[attrNetworkType].GetStringValue()) + require.Equal(t, int64(4200), a["flow.io.bytes"].GetIntValue()) + require.Equal(t, int64(42), a["flow.io.packets"].GetIntValue()) + require.Equal(t, int64(64500), a["flow.src_as"].GetIntValue()) + require.Equal(t, "10.0.0.254", a["flow.next_hop"].GetStringValue()) + require.Equal(t, uint64(time.Unix(1700000000, 0).UnixNano()), rec.TimeUnixNano) +} + +func TestTransportName(t *testing.T) { + require.Equal(t, "tcp", transportName(6)) + require.Equal(t, "udp", transportName(17)) + require.Equal(t, "icmp", transportName(1)) + require.Equal(t, "unknown", transportName(99)) +} diff --git a/embed/record.go b/embed/record.go index ee86e5e0..8287b67f 100644 --- a/embed/record.go +++ b/embed/record.go @@ -1,6 +1,9 @@ package embed -import "time" +import ( + "net" + "time" +) // LogRecord represents a single log entry. type LogRecord struct { @@ -158,3 +161,57 @@ type SpanMetadata struct { Attributes map[string]any Resource map[string]any } + +// FlowRecord represents a single network flow record — one observed +// conversation identified by the classic 5-tuple, plus the byte/packet +// counters and routing dimensions the NetFlow/IPFIX/sFlow wire encoders +// carry. It is blitz's fourth signal type, alongside LogRecord, +// MetricPoint, and Span. +// +// Addresses are net.IP so a single record serves both IPv4 and IPv6 +// encoders. StartTime/EndTime are the flow's first/last packet times; +// encoders that use switch-relative timestamps (NetFlow v5/v9) derive +// them against a boot time the encoder owns. +type FlowRecord struct { + // SrcIP, DstIP are the flow endpoints. + SrcIP, DstIP net.IP + // SrcPort, DstPort are the transport ports (0 for non-port protocols). + SrcPort, DstPort uint16 + // Protocol is the IP protocol number (6 TCP, 17 UDP, 1 ICMP, ...). + Protocol uint8 + // TOS is the IP type-of-service / DSCP byte. + TOS uint8 + // TCPFlags is the cumulative OR of TCP flags across the flow. + TCPFlags uint8 + + // NextHop is the IP of the next-hop router (may be nil). + NextHop net.IP + // InputIface, OutputIface are SNMP ifIndex values. + InputIface, OutputIface uint32 + + // Packets, Bytes are the flow's packet and byte counts. + Packets, Bytes uint64 + + // StartTime, EndTime are the first and last packet timestamps. + StartTime, EndTime time.Time + + // SrcAS, DstAS are the source/destination autonomous-system numbers. + SrcAS, DstAS uint32 + // SrcMask, DstMask are the source/destination prefix lengths. + SrcMask, DstMask uint8 + + // SamplingRate is the packet sampling rate (sFlow; 0 = unsampled). + SamplingRate uint32 + + // Metadata carries timestamp, per-record Attributes, and Resource for + // the OTel/log representation. See LogRecordMetadata for conventions. + Metadata FlowRecordMetadata +} + +// FlowRecordMetadata is the metadata for a flow record. See +// LogRecordMetadata's doc comment for Resource/Attributes semantics. +type FlowRecordMetadata struct { + Timestamp time.Time + Attributes map[string]any + Resource map[string]any +} diff --git a/embed/runner.go b/embed/runner.go index 570b6d98..e1a5bcf1 100644 --- a/embed/runner.go +++ b/embed/runner.go @@ -52,9 +52,10 @@ type Config struct { // Modules lists the ProducerModules the Runner will operate. Modules []ProducerModule - // Logs, Metrics, Traces configure backpressure per signal channel. + // Logs, Metrics, Traces, Flows configure backpressure per signal channel. // Zero values mean BackpressureBlock. Logs ConsumerBackpressure Metrics ConsumerBackpressure Traces ConsumerBackpressure + Flows ConsumerBackpressure } diff --git a/flow/ipfix/ipfix.go b/flow/ipfix/ipfix.go new file mode 100644 index 00000000..c6d5fa83 --- /dev/null +++ b/flow/ipfix/ipfix.go @@ -0,0 +1,240 @@ +// Package ipfix encodes blitz FlowRecords into IPFIX (RFC 7011) export +// messages. IPFIX is the IETF standardization of NetFlow v9: template-based, +// but with a 16-byte header, absolute millisecond timestamps, and 64-bit +// counters. +// +// A Vendor makes the export genuinely vendor-distinct on the wire: IPFIX is the +// only format that carries an enterprise-specific Information Element (a field +// whose type has the 0x8000 bit set, followed by a 4-byte Private Enterprise +// Number). A decoder such as netsampler/goflow2 reads that PEN back +// (DataField.PenProvided / .Pen), so an AppFlow (Citrix, PEN 5951) packet +// decodes distinctly from a jFlow (Juniper, 2636) or cflowd (Nokia, 6527) one. +package ipfix + +import ( + "encoding/binary" + "net" + "sync" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/flow/wire" +) + +const ( + headerLen = 16 + templateID uint16 = 256 + setIDTemplate uint16 = 2 + + // enterpriseBit marks a field type as enterprise-specific; the 4-byte PEN + // follows the field in the template (RFC 7011 §3.2). + enterpriseBit uint16 = 0x8000 + + // DefaultRefreshInterval and DefaultRefreshRecords are the template resend + // cadence: whichever comes first. + DefaultRefreshInterval = 60 * time.Second + DefaultRefreshRecords = 256 +) + +// Vendor is an enterprise-specific IE that stamps the export as a vendor's +// IPFIX derivative. PEN is the IANA Private Enterprise Number; IEID is the +// vendor's element id (without the enterprise bit). +type Vendor struct { + PEN uint32 + IEID uint16 +} + +type ie struct { + id uint16 + len uint16 +} + +// template is the fixed IPv4 IPFIX template (IANA information elements). +var template = []ie{ + {1, 8}, // octetDeltaCount + {2, 8}, // packetDeltaCount + {4, 1}, // protocolIdentifier + {5, 1}, // ipClassOfService + {6, 1}, // tcpControlBits + {7, 2}, // sourceTransportPort + {8, 4}, // sourceIPv4Address + {9, 1}, // sourceIPv4PrefixLength + {10, 4}, // ingressInterface + {11, 2}, // destinationTransportPort + {12, 4}, // destinationIPv4Address + {13, 1}, // destinationIPv4PrefixLength + {14, 4}, // egressInterface + {15, 4}, // ipNextHopIPv4Address + {16, 4}, // bgpSourceAsNumber + {17, 4}, // bgpDestinationAsNumber + {152, 8}, // flowStartMilliseconds + {153, 8}, // flowEndMilliseconds +} + +// vendorIELen is the byte width of the vendor enterprise IE value. +const vendorIELen = 4 + +func recordLen(vendor *Vendor) int { + n := 0 + for _, f := range template { + n += int(f.len) + } + if vendor != nil { + n += vendorIELen + } + return n +} + +// Encoder encodes FlowRecords as IPFIX messages with periodic template +// resends. Safe for concurrent Encode calls. +type Encoder struct { + domainID uint32 + vendor *Vendor + + refreshInterval time.Duration + refreshRecords int + + mu sync.Mutex + seqRecords uint32 // data records sent so far = next message's sequence number + lastTemplate time.Time + firstDone bool +} + +// NewEncoder returns an IPFIX encoder. A non-nil vendor adds that vendor's +// enterprise IE to every template and record. +func NewEncoder(domainID uint32, vendor *Vendor) *Encoder { + return &Encoder{ + domainID: domainID, + vendor: vendor, + refreshInterval: DefaultRefreshInterval, + refreshRecords: DefaultRefreshRecords, + } +} + +// Encode renders flows into a single IPFIX message, prepending a template set +// when a resend is due. Returns nil for an empty batch. +func (e *Encoder) Encode(now time.Time, flows []embed.FlowRecord) [][]byte { + if len(flows) == 0 { + return nil + } + e.mu.Lock() + defer e.mu.Unlock() + + cumBefore := wire.U64(e.seqRecords) + cumAfter := cumBefore + wire.U64(len(flows)) + withTemplate := !e.firstDone || + (e.refreshInterval > 0 && !e.lastTemplate.IsZero() && now.Sub(e.lastTemplate) >= e.refreshInterval) || + (e.refreshRecords > 0 && cumBefore/wire.U64(e.refreshRecords) != cumAfter/wire.U64(e.refreshRecords)) + + recLen := recordLen(e.vendor) + dataSetLen := 4 + len(flows)*recLen + if pad := dataSetLen % 4; pad != 0 { + dataSetLen += 4 - pad + } + tmplSetLen := 0 + if withTemplate { + tmplSetLen = e.templateSetLen() + e.lastTemplate = now + } + + buf := make([]byte, headerLen+tmplSetLen+dataSetLen) + binary.BigEndian.PutUint16(buf[0:2], 10) + binary.BigEndian.PutUint16(buf[2:4], wire.U16(len(buf))) + binary.BigEndian.PutUint32(buf[4:8], wire.U32(now.Unix())) + binary.BigEndian.PutUint32(buf[8:12], e.seqRecords) + binary.BigEndian.PutUint32(buf[12:16], e.domainID) + + off := headerLen + if withTemplate { + off += e.putTemplateSet(buf[off:]) + } + e.putDataSet(buf[off:], flows, recLen, dataSetLen) + + e.firstDone = true + e.seqRecords = wire.U32(cumAfter) + return [][]byte{buf} +} + +// templateSetLen is the byte length of the template set including the optional +// vendor enterprise IE (which carries a trailing 4-byte PEN). +func (e *Encoder) templateSetLen() int { + n := 4 + 4 + len(template)*4 + if e.vendor != nil { + n += 4 + 4 // enterprise IE: (type,len) + PEN + } + return n +} + +func (e *Encoder) putTemplateSet(b []byte) int { + total := e.templateSetLen() + fieldCount := len(template) + if e.vendor != nil { + fieldCount++ + } + binary.BigEndian.PutUint16(b[0:2], setIDTemplate) + binary.BigEndian.PutUint16(b[2:4], wire.U16(total)) + binary.BigEndian.PutUint16(b[4:6], templateID) + binary.BigEndian.PutUint16(b[6:8], wire.U16(fieldCount)) + off := 8 + for _, f := range template { + binary.BigEndian.PutUint16(b[off:off+2], f.id) + binary.BigEndian.PutUint16(b[off+2:off+4], f.len) + off += 4 + } + if e.vendor != nil { + binary.BigEndian.PutUint16(b[off:off+2], e.vendor.IEID|enterpriseBit) + binary.BigEndian.PutUint16(b[off+2:off+4], vendorIELen) + binary.BigEndian.PutUint32(b[off+4:off+8], e.vendor.PEN) + off += 8 + } + return total +} + +func (e *Encoder) putDataSet(b []byte, flows []embed.FlowRecord, recLen, setLen int) { + binary.BigEndian.PutUint16(b[0:2], templateID) + binary.BigEndian.PutUint16(b[2:4], wire.U16(setLen)) + off := 4 + for _, f := range flows { + e.putRecord(b[off:off+recLen], f) + off += recLen + } +} + +func (e *Encoder) putRecord(b []byte, f embed.FlowRecord) { + binary.BigEndian.PutUint64(b[0:8], f.Bytes) + binary.BigEndian.PutUint64(b[8:16], f.Packets) + b[16] = f.Protocol + b[17] = f.TOS + b[18] = f.TCPFlags + binary.BigEndian.PutUint16(b[19:21], f.SrcPort) + copy(b[21:25], ip4(f.SrcIP)) + b[25] = f.SrcMask + binary.BigEndian.PutUint32(b[26:30], f.InputIface) + binary.BigEndian.PutUint16(b[30:32], f.DstPort) + copy(b[32:36], ip4(f.DstIP)) + b[36] = f.DstMask + binary.BigEndian.PutUint32(b[37:41], f.OutputIface) + copy(b[41:45], ip4(f.NextHop)) + binary.BigEndian.PutUint32(b[45:49], f.SrcAS) + binary.BigEndian.PutUint32(b[49:53], f.DstAS) + binary.BigEndian.PutUint64(b[53:61], msSinceEpoch(f.StartTime)) + binary.BigEndian.PutUint64(b[61:69], msSinceEpoch(f.EndTime)) + if e.vendor != nil { + // Vendor IE value: the PEN echoed, so the decoded value is self-describing. + binary.BigEndian.PutUint32(b[69:73], e.vendor.PEN) + } +} + +func msSinceEpoch(t time.Time) uint64 { + if t.IsZero() { + return 0 + } + return wire.U64(t.UnixMilli()) +} + +func ip4(ip net.IP) []byte { + if v4 := ip.To4(); v4 != nil { + return v4 + } + return []byte{0, 0, 0, 0} +} diff --git a/flow/ipfix/ipfix_test.go b/flow/ipfix/ipfix_test.go new file mode 100644 index 00000000..54ffc212 --- /dev/null +++ b/flow/ipfix/ipfix_test.go @@ -0,0 +1,50 @@ +package ipfix + +import ( + "encoding/binary" + "net" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" +) + +func testFlow() embed.FlowRecord { + return embed.FlowRecord{ + SrcIP: net.IPv4(10, 0, 0, 1), DstIP: net.IPv4(93, 184, 216, 34), + SrcPort: 54321, DstPort: 443, Protocol: 6, TCPFlags: 0x1b, + NextHop: net.IPv4(10, 0, 0, 254), InputIface: 2, OutputIface: 3, + Packets: 42, Bytes: 4200, SrcAS: 64500, DstAS: 15133, SrcMask: 24, DstMask: 19, + StartTime: time.Unix(1700000000, 0), EndTime: time.Unix(1700000005, 0), + } +} + +func TestIPFIXFirstMessageHasTemplateAndData(t *testing.T) { + enc := NewEncoder(0, nil) + pkts := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()}) + require.Len(t, pkts, 1) + msg := pkts[0] + + require.Equal(t, uint16(10), binary.BigEndian.Uint16(msg[0:2])) // version + require.Equal(t, uint16(len(msg)), binary.BigEndian.Uint16(msg[2:4])) // length = whole message + require.Equal(t, uint32(1700000010), binary.BigEndian.Uint32(msg[4:8])) // export time + require.Equal(t, uint32(0), binary.BigEndian.Uint32(msg[8:12])) // sequence (data records prior) + + set := msg[16:] + require.Equal(t, uint16(2), binary.BigEndian.Uint16(set[0:2])) // set_id 2 = template + setLen := binary.BigEndian.Uint16(set[2:4]) + require.Equal(t, templateID, binary.BigEndian.Uint16(set[4:6])) + + data := set[setLen:] + require.Equal(t, templateID, binary.BigEndian.Uint16(data[0:2])) // data set id = template id +} + +func TestIPFIXSequenceCountsDataRecords(t *testing.T) { + enc := NewEncoder(0, nil) + now := time.Unix(1700000010, 0) + enc.Encode(now, []embed.FlowRecord{testFlow(), testFlow()}) // 2 data records + msg := enc.Encode(now, []embed.FlowRecord{testFlow()})[0] + // sequence = number of data records sent before this message = 2 + require.Equal(t, uint32(2), binary.BigEndian.Uint32(msg[8:12])) +} diff --git a/flow/ipfix/roundtrip_test.go b/flow/ipfix/roundtrip_test.go new file mode 100644 index 00000000..79df538e --- /dev/null +++ b/flow/ipfix/roundtrip_test.go @@ -0,0 +1,85 @@ +package ipfix + +import ( + "bytes" + "encoding/binary" + "testing" + "time" + + nf "github.com/netsampler/goflow2/v2/decoders/netflow" + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" +) + +// decodeIPFIX decodes an IPFIX message with goflow2 and returns the data records. +func decodeIPFIX(t *testing.T, msg []byte) []nf.DataRecord { + t.Helper() + ts := nf.CreateTemplateSystem() + var v9 nf.NFv9Packet + var pkt nf.IPFIXPacket + require.NoError(t, nf.DecodeMessageVersion(bytes.NewBuffer(msg), ts, &v9, &pkt)) + require.Equal(t, uint16(10), pkt.Version) + var recs []nf.DataRecord + for _, fs := range pkt.FlowSets { + if d, ok := fs.(nf.DataFlowSet); ok { + recs = append(recs, d.Records...) + } + } + return recs +} + +func field(rec nf.DataRecord, typ uint16) ([]byte, bool) { + for _, v := range rec.Values { + if v.Type == typ { + b, _ := v.Value.([]byte) + return b, true + } + } + return nil, false +} + +func TestIPFIXRoundTripGoflow2(t *testing.T) { + enc := NewEncoder(0, nil) + msg := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()})[0] + recs := decodeIPFIX(t, msg) + require.Len(t, recs, 1) + + // octetDeltaCount (IE 1) round-trips to 4200 (64-bit in IPFIX). + b, ok := field(recs[0], 1) + require.True(t, ok) + require.Equal(t, uint64(4200), binary.BigEndian.Uint64(b)) + // sourceTransportPort (IE 7) round-trips to 54321. + sp, ok := field(recs[0], 7) + require.True(t, ok) + require.Equal(t, uint16(54321), binary.BigEndian.Uint16(sp)) +} + +func TestIPFIXVendorRoundTripGoflow2(t *testing.T) { + cases := []struct { + name string + pen uint32 + ieid uint16 + }{ + {"appflow-citrix", 5951, 130}, + {"jflow-juniper", 2636, 137}, + {"cflowd-nokia", 6527, 91}, + {"netstream-huawei", 2011, 140}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + enc := NewEncoder(0, &Vendor{PEN: tc.pen, IEID: tc.ieid}) + msg := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()})[0] + recs := decodeIPFIX(t, msg) + require.Len(t, recs, 1) + + var found bool + for _, v := range recs[0].Values { + if v.PenProvided && v.Pen == tc.pen { + found = true + require.Equal(t, tc.ieid, v.Type, "enterprise IE id") + } + } + require.True(t, found, "decoded record must carry enterprise PEN %d", tc.pen) + }) + } +} diff --git a/flow/netflow/roundtrip_test.go b/flow/netflow/roundtrip_test.go new file mode 100644 index 00000000..3fe67028 --- /dev/null +++ b/flow/netflow/roundtrip_test.go @@ -0,0 +1,97 @@ +package netflow + +import ( + "bytes" + "encoding/binary" + "testing" + "time" + + nf "github.com/netsampler/goflow2/v2/decoders/netflow" + nfl "github.com/netsampler/goflow2/v2/decoders/netflowlegacy" + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" +) + +// TestV5RoundTripGoflow2 encodes with blitz and decodes with the exact decoder +// the shipped collector runs (netsampler/goflow2 netflowlegacy), asserting the +// decoded flow matches the input. +func TestV5RoundTripGoflow2(t *testing.T) { + enc := NewV5Encoder(time.Unix(1699999000, 0)) + pkt := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()})[0] + + var p nfl.PacketNetFlowV5 + require.NoError(t, nfl.DecodeMessageVersion(bytes.NewBuffer(pkt), &p)) + require.Equal(t, uint16(5), p.Version) + require.Len(t, p.Records, 1) + r := p.Records[0] + require.Equal(t, uint32(0x0A000001), uint32(r.SrcAddr)) // 10.0.0.1 + require.Equal(t, uint16(443), r.DstPort) + require.Equal(t, uint32(42), r.DPkts) + require.Equal(t, uint32(4200), r.DOctets) + require.Equal(t, uint8(6), r.Proto) +} + +// decodeV9 decodes a v9 packet with goflow2 and returns the data records. +func decodeV9(t *testing.T, pkt []byte) []nf.DataRecord { + t.Helper() + ts := nf.CreateTemplateSystem() + var v9 nf.NFv9Packet + var ipfix nf.IPFIXPacket + require.NoError(t, nf.DecodeMessageVersion(bytes.NewBuffer(pkt), ts, &v9, &ipfix)) + require.Equal(t, uint16(9), v9.Version) + var recs []nf.DataRecord + for _, fs := range v9.FlowSets { + if d, ok := fs.(nf.DataFlowSet); ok { + recs = append(recs, d.Records...) + } + } + return recs +} + +func fieldValue(rec nf.DataRecord, typ uint16) ([]byte, bool) { + for _, v := range rec.Values { + if v.Type == typ { + b, _ := v.Value.([]byte) + return b, true + } + } + return nil, false +} + +func TestV9RoundTripGoflow2(t *testing.T) { + enc := NewV9Encoder(time.Unix(1699999000, 0), 0, 0) + pkt := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()})[0] + recs := decodeV9(t, pkt) + require.Len(t, recs, 1) + + // IN_BYTES (type 1) round-trips to 4200. + b, ok := fieldValue(recs[0], 1) + require.True(t, ok) + require.Equal(t, uint32(4200), binary.BigEndian.Uint32(b)) + // L4_SRC_PORT (type 7) round-trips to 54321. + sp, ok := fieldValue(recs[0], 7) + require.True(t, ok) + require.Equal(t, uint16(54321), binary.BigEndian.Uint16(sp)) +} + +func TestV9VendorRoundTripGoflow2(t *testing.T) { + cases := []struct { + name string + field uint16 + }{ + {"netstream-huawei", 0x9001}, + {"rflow-redback", 0x9002}, + {"jflow-juniper", 0x9003}, + {"cflowd-nokia", 0x9004}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + enc := NewV9Encoder(time.Unix(1699999000, 0), 0, tc.field) + pkt := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()})[0] + recs := decodeV9(t, pkt) + require.Len(t, recs, 1) + _, ok := fieldValue(recs[0], tc.field) + require.True(t, ok, "decoded record must carry the vendor field type 0x%x", tc.field) + }) + } +} diff --git a/flow/netflow/v5.go b/flow/netflow/v5.go new file mode 100644 index 00000000..5a2c8c4e --- /dev/null +++ b/flow/netflow/v5.go @@ -0,0 +1,114 @@ +// Package netflow encodes blitz FlowRecords into Cisco NetFlow export +// packets. V5 is the fixed-layout legacy format (IPv4 only, 30 flows per +// packet); V9 is the template-based format (see v9.go). +package netflow + +import ( + "encoding/binary" + "net" + "sync/atomic" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/flow/wire" +) + +const ( + v5HeaderLen = 24 + v5RecordLen = 48 + v5MaxFlows = 30 +) + +// V5Encoder encodes FlowRecords as NetFlow v5 packets. It owns the export +// boot time (for switch-uptime timestamps) and a flow sequence counter. +// Safe for concurrent Encode calls. +type V5Encoder struct { + boot time.Time + seq atomic.Uint32 + engineType uint8 + engineID uint8 + sampling uint16 +} + +// NewV5Encoder returns a v5 encoder whose switch-uptime clock starts at boot. +func NewV5Encoder(boot time.Time) *V5Encoder { + return &V5Encoder{boot: boot} +} + +// Encode renders flows into one or more NetFlow v5 packets (max 30 flows +// each), stamped at now. Returns nil for an empty batch. +func (e *V5Encoder) Encode(now time.Time, flows []embed.FlowRecord) [][]byte { + if len(flows) == 0 { + return nil + } + var pkts [][]byte + for start := 0; start < len(flows); start += v5MaxFlows { + end := start + v5MaxFlows + if end > len(flows) { + end = len(flows) + } + pkts = append(pkts, e.packet(now, flows[start:end])) + } + return pkts +} + +func (e *V5Encoder) packet(now time.Time, flows []embed.FlowRecord) []byte { + count := len(flows) + buf := make([]byte, v5HeaderLen+count*v5RecordLen) + + uptimeMS := wire.U32(now.Sub(e.boot).Milliseconds()) + seq := e.seq.Add(wire.U32(count)) - wire.U32(count) // sequence of first record + + binary.BigEndian.PutUint16(buf[0:2], 5) + binary.BigEndian.PutUint16(buf[2:4], wire.U16(count)) + binary.BigEndian.PutUint32(buf[4:8], uptimeMS) + binary.BigEndian.PutUint32(buf[8:12], wire.U32(now.Unix())) + binary.BigEndian.PutUint32(buf[12:16], wire.U32(now.Nanosecond())) + binary.BigEndian.PutUint32(buf[16:20], seq) + buf[20] = e.engineType + buf[21] = e.engineID + binary.BigEndian.PutUint16(buf[22:24], e.sampling) + + for i, f := range flows { + e.putRecord(buf[v5HeaderLen+i*v5RecordLen:], f) + } + return buf +} + +func (e *V5Encoder) putRecord(b []byte, f embed.FlowRecord) { + copy(b[0:4], ip4(f.SrcIP)) + copy(b[4:8], ip4(f.DstIP)) + copy(b[8:12], ip4(f.NextHop)) + binary.BigEndian.PutUint16(b[12:14], wire.U16(f.InputIface)) + binary.BigEndian.PutUint16(b[14:16], wire.U16(f.OutputIface)) + binary.BigEndian.PutUint32(b[16:20], wire.U32(f.Packets)) + binary.BigEndian.PutUint32(b[20:24], wire.U32(f.Bytes)) + binary.BigEndian.PutUint32(b[24:28], e.uptimeMS(f.StartTime)) + binary.BigEndian.PutUint32(b[28:32], e.uptimeMS(f.EndTime)) + binary.BigEndian.PutUint16(b[32:34], f.SrcPort) + binary.BigEndian.PutUint16(b[34:36], f.DstPort) + // b[36] pad1 + b[37] = f.TCPFlags + b[38] = f.Protocol + b[39] = f.TOS + binary.BigEndian.PutUint16(b[40:42], wire.U16(f.SrcAS)) + binary.BigEndian.PutUint16(b[42:44], wire.U16(f.DstAS)) + b[44] = f.SrcMask + b[45] = f.DstMask + // b[46:48] pad2 +} + +func (e *V5Encoder) uptimeMS(t time.Time) uint32 { + if t.IsZero() { + return 0 + } + return wire.U32(t.Sub(e.boot).Milliseconds()) +} + +// ip4 returns the 4-byte IPv4 form of ip, or zeros if ip is nil/not v4. +func ip4(ip net.IP) []byte { + if v4 := ip.To4(); v4 != nil { + return v4 + } + return []byte{0, 0, 0, 0} +} diff --git a/flow/netflow/v5_test.go b/flow/netflow/v5_test.go new file mode 100644 index 00000000..491b9d66 --- /dev/null +++ b/flow/netflow/v5_test.go @@ -0,0 +1,94 @@ +package netflow + +import ( + "encoding/binary" + "net" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" +) + +func testFlow() embed.FlowRecord { + return embed.FlowRecord{ + SrcIP: net.IPv4(10, 0, 0, 1), + DstIP: net.IPv4(93, 184, 216, 34), + SrcPort: 54321, + DstPort: 443, + Protocol: 6, + TOS: 0, + TCPFlags: 0x1b, + NextHop: net.IPv4(10, 0, 0, 254), + InputIface: 2, + OutputIface: 3, + Packets: 42, + Bytes: 4200, + SrcAS: 64500, + DstAS: 15133, + SrcMask: 24, + DstMask: 19, + StartTime: time.Unix(1700000000, 0), + EndTime: time.Unix(1700000005, 0), + } +} + +func TestV5EncodeHeaderAndRecord(t *testing.T) { + boot := time.Unix(1699999000, 0) + enc := NewV5Encoder(boot) + now := time.Unix(1700000010, 0) + + pkts := enc.Encode(now, []embed.FlowRecord{testFlow()}) + require.Len(t, pkts, 1) + pkt := pkts[0] + require.Len(t, pkt, 24+48) // header + one record + + require.Equal(t, uint16(5), binary.BigEndian.Uint16(pkt[0:2])) // version + require.Equal(t, uint16(1), binary.BigEndian.Uint16(pkt[2:4])) // count + require.Equal(t, uint32(1010*1000), binary.BigEndian.Uint32(pkt[4:8])) // sysuptime ms (now-boot) + require.Equal(t, uint32(1700000010), binary.BigEndian.Uint32(pkt[8:12])) // unix_secs + + rec := pkt[24:] + require.Equal(t, []byte{10, 0, 0, 1}, rec[0:4]) // srcaddr + require.Equal(t, []byte{93, 184, 216, 34}, rec[4:8]) // dstaddr + require.Equal(t, []byte{10, 0, 0, 254}, rec[8:12]) // nexthop + require.Equal(t, uint16(2), binary.BigEndian.Uint16(rec[12:14])) // input + require.Equal(t, uint16(3), binary.BigEndian.Uint16(rec[14:16])) // output + require.Equal(t, uint32(42), binary.BigEndian.Uint32(rec[16:20])) // dPkts + require.Equal(t, uint32(4200), binary.BigEndian.Uint32(rec[20:24])) // dOctets + require.Equal(t, uint16(54321), binary.BigEndian.Uint16(rec[32:34])) // srcport + require.Equal(t, uint16(443), binary.BigEndian.Uint16(rec[34:36])) // dstport + require.Equal(t, uint8(0x1b), rec[37]) // tcp_flags + require.Equal(t, uint8(6), rec[38]) // prot + require.Equal(t, uint16(64500), binary.BigEndian.Uint16(rec[40:42])) // src_as + require.Equal(t, uint16(15133), binary.BigEndian.Uint16(rec[42:44])) // dst_as + require.Equal(t, uint8(24), rec[44]) // src_mask + require.Equal(t, uint8(19), rec[45]) // dst_mask + + // first/last are sysuptime-relative ms + first := binary.BigEndian.Uint32(rec[24:28]) + last := binary.BigEndian.Uint32(rec[28:32]) + require.Equal(t, uint32((1700000000-1699999000)*1000), first) + require.Equal(t, uint32((1700000005-1699999000)*1000), last) +} + +func TestV5SplitsAt30Records(t *testing.T) { + enc := NewV5Encoder(time.Unix(1699999000, 0)) + flows := make([]embed.FlowRecord, 31) + for i := range flows { + flows[i] = testFlow() + } + pkts := enc.Encode(time.Unix(1700000010, 0), flows) + require.Len(t, pkts, 2) + require.Equal(t, uint16(30), binary.BigEndian.Uint16(pkts[0][2:4])) + require.Equal(t, uint16(1), binary.BigEndian.Uint16(pkts[1][2:4])) +} + +func TestV5FlowSequenceAdvances(t *testing.T) { + enc := NewV5Encoder(time.Unix(1699999000, 0)) + now := time.Unix(1700000010, 0) + enc.Encode(now, []embed.FlowRecord{testFlow(), testFlow()}) + pkts := enc.Encode(now, []embed.FlowRecord{testFlow()}) + // flow_sequence counts records, so the second packet starts at 2. + require.Equal(t, uint32(2), binary.BigEndian.Uint32(pkts[0][16:20])) +} diff --git a/flow/netflow/v9.go b/flow/netflow/v9.go new file mode 100644 index 00000000..5e29ed00 --- /dev/null +++ b/flow/netflow/v9.go @@ -0,0 +1,232 @@ +package netflow + +import ( + "encoding/binary" + "sync" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/flow/wire" +) + +const ( + v9HeaderLen = 20 + templateID uint16 = 256 + + // DefaultRefreshInterval and DefaultRefreshRecords are the NetFlow v9 / + // IPFIX template resend cadence: whichever comes first. + DefaultRefreshInterval = 60 * time.Second + DefaultRefreshRecords = 256 +) + +// v9Field is one template field: NetFlow field type and its byte length. +type v9Field struct { + typ uint16 + len uint16 +} + +// v9Template is the fixed IPv4 template both the v9 and IPFIX encoders share +// (IPFIX reuses the same field type IDs). Order defines the data-record layout. +var v9Template = []v9Field{ + {1, 4}, // IN_BYTES + {2, 4}, // IN_PKTS + {4, 1}, // PROTOCOL + {5, 1}, // TOS + {6, 1}, // TCP_FLAGS + {7, 2}, // L4_SRC_PORT + {8, 4}, // IPV4_SRC_ADDR + {9, 1}, // SRC_MASK + {10, 2}, // INPUT_SNMP + {11, 2}, // L4_DST_PORT + {12, 4}, // IPV4_DST_ADDR + {13, 1}, // DST_MASK + {14, 2}, // OUTPUT_SNMP + {15, 4}, // IPV4_NEXT_HOP + {16, 2}, // SRC_AS + {17, 2}, // DST_AS + {21, 4}, // LAST_SWITCHED + {22, 4}, // FIRST_SWITCHED +} + +func recordLen(tmpl []v9Field) int { + n := 0 + for _, f := range tmpl { + n += int(f.len) + } + return n +} + +// V9Encoder encodes FlowRecords as NetFlow v9 packets with periodic template +// resends. Safe for concurrent Encode calls. +// +// NetFlow v9 has no enterprise-PEN mechanism (that is IPFIX-only), so a v9 +// vendor flavor is distinguished by a proprietary field type in the template: +// vendorField, when non-zero, appends one vendor-specific 4-byte field that a +// decoder surfaces by its type id. This is the genuine wire difference for +// v9-based flavors such as NetStream (Huawei) and rFlow (Redback). +type V9Encoder struct { + boot time.Time + sourceID uint32 + vendorField uint16 + + refreshInterval time.Duration + refreshRecords int + + mu sync.Mutex + pktSeq uint32 + totalRecords uint64 + lastTemplate time.Time +} + +// NewV9Encoder returns a v9 encoder with the default template refresh cadence. +// A non-zero vendorField appends that vendor-specific field to every template +// and record. +func NewV9Encoder(boot time.Time, sourceID uint32, vendorField uint16) *V9Encoder { + return &V9Encoder{ + boot: boot, + sourceID: sourceID, + vendorField: vendorField, + refreshInterval: DefaultRefreshInterval, + refreshRecords: DefaultRefreshRecords, + } +} + +// vendorLen is the byte width of the v9 vendor-specific field value. +const vendorLen = 4 + +func (e *V9Encoder) recLen() int { + n := recordLen(v9Template) + if e.vendorField != 0 { + n += vendorLen + } + return n +} + +func (e *V9Encoder) templateLen() int { + n := 4 + 4 + len(v9Template)*4 + if e.vendorField != 0 { + n += 4 + } + return n +} + +// Encode renders flows into a single NetFlow v9 packet, prepending a template +// flowset when a resend is due (first packet, interval elapsed, or the record +// counter crossed a refresh boundary). Returns nil for an empty batch. +func (e *V9Encoder) Encode(now time.Time, flows []embed.FlowRecord) [][]byte { + if len(flows) == 0 { + return nil + } + e.mu.Lock() + defer e.mu.Unlock() + + cumBefore := e.totalRecords + cumAfter := cumBefore + wire.U64(len(flows)) + withTemplate := e.pktSeq == 0 || + (e.refreshInterval > 0 && !e.lastTemplate.IsZero() && now.Sub(e.lastTemplate) >= e.refreshInterval) || + (e.refreshRecords > 0 && cumBefore/wire.U64(e.refreshRecords) != cumAfter/wire.U64(e.refreshRecords)) + + recLen := e.recLen() + dataLen := 4 + len(flows)*recLen + if pad := dataLen % 4; pad != 0 { + dataLen += 4 - pad + } + tmplLen := 0 + if withTemplate { + tmplLen = e.templateLen() // flowset hdr + template hdr + fields (+ vendor field) + e.lastTemplate = now + } + + buf := make([]byte, v9HeaderLen+tmplLen+dataLen) + records := len(flows) + if withTemplate { + records++ + } + + binary.BigEndian.PutUint16(buf[0:2], 9) + binary.BigEndian.PutUint16(buf[2:4], wire.U16(records)) + binary.BigEndian.PutUint32(buf[4:8], wire.U32(now.Sub(e.boot).Milliseconds())) + binary.BigEndian.PutUint32(buf[8:12], wire.U32(now.Unix())) + binary.BigEndian.PutUint32(buf[12:16], e.pktSeq) + binary.BigEndian.PutUint32(buf[16:20], e.sourceID) + + off := v9HeaderLen + if withTemplate { + off += e.putV9Template(buf[off:]) + } + e.putV9DataFlowSet(buf[off:], flows, recLen, dataLen) + + e.pktSeq++ + e.totalRecords = cumAfter + return [][]byte{buf} +} + +// putV9Template writes the template flowset and returns its length. +func (e *V9Encoder) putV9Template(b []byte) int { + total := e.templateLen() + fieldCount := len(v9Template) + if e.vendorField != 0 { + fieldCount++ + } + binary.BigEndian.PutUint16(b[0:2], 0) // flowset_id 0 = template + binary.BigEndian.PutUint16(b[2:4], wire.U16(total)) + binary.BigEndian.PutUint16(b[4:6], templateID) + binary.BigEndian.PutUint16(b[6:8], wire.U16(fieldCount)) + off := 8 + for _, f := range v9Template { + binary.BigEndian.PutUint16(b[off:off+2], f.typ) + binary.BigEndian.PutUint16(b[off+2:off+4], f.len) + off += 4 + } + if e.vendorField != 0 { + binary.BigEndian.PutUint16(b[off:off+2], e.vendorField) + binary.BigEndian.PutUint16(b[off+2:off+4], vendorLen) + off += 4 + } + return total +} + +func (e *V9Encoder) putV9DataFlowSet(b []byte, flows []embed.FlowRecord, recLen, dataLen int) { + binary.BigEndian.PutUint16(b[0:2], templateID) + binary.BigEndian.PutUint16(b[2:4], wire.U16(dataLen)) + off := 4 + for _, f := range flows { + e.putV9Record(b[off:off+recLen], f) + off += recLen + } + // remaining bytes stay zero = padding +} + +func (e *V9Encoder) putV9Record(b []byte, f embed.FlowRecord) { + boot := e.boot + binary.BigEndian.PutUint32(b[0:4], wire.U32(f.Bytes)) + binary.BigEndian.PutUint32(b[4:8], wire.U32(f.Packets)) + b[8] = f.Protocol + b[9] = f.TOS + b[10] = f.TCPFlags + binary.BigEndian.PutUint16(b[11:13], f.SrcPort) + copy(b[13:17], ip4(f.SrcIP)) + b[17] = f.SrcMask + binary.BigEndian.PutUint16(b[18:20], wire.U16(f.InputIface)) + binary.BigEndian.PutUint16(b[20:22], f.DstPort) + copy(b[22:26], ip4(f.DstIP)) + b[26] = f.DstMask + binary.BigEndian.PutUint16(b[27:29], wire.U16(f.OutputIface)) + copy(b[29:33], ip4(f.NextHop)) + binary.BigEndian.PutUint16(b[33:35], wire.U16(f.SrcAS)) + binary.BigEndian.PutUint16(b[35:37], wire.U16(f.DstAS)) + binary.BigEndian.PutUint32(b[37:41], uptime(boot, f.EndTime)) + binary.BigEndian.PutUint32(b[41:45], uptime(boot, f.StartTime)) + if e.vendorField != 0 { + // Vendor field value: the vendor field id echoed, so the record is + // self-describing and the field's presence is verifiable. + binary.BigEndian.PutUint32(b[45:49], wire.U32(e.vendorField)) + } +} + +func uptime(boot, t time.Time) uint32 { + if t.IsZero() { + return 0 + } + return wire.U32(t.Sub(boot).Milliseconds()) +} diff --git a/flow/netflow/v9_test.go b/flow/netflow/v9_test.go new file mode 100644 index 00000000..8b5d52e4 --- /dev/null +++ b/flow/netflow/v9_test.go @@ -0,0 +1,53 @@ +package netflow + +import ( + "encoding/binary" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" +) + +func TestV9FirstPacketCarriesTemplateAndData(t *testing.T) { + enc := NewV9Encoder(time.Unix(1699999000, 0), 0, 0) + pkts := enc.Encode(time.Unix(1700000010, 0), []embed.FlowRecord{testFlow()}) + require.Len(t, pkts, 1) + pkt := pkts[0] + + require.Equal(t, uint16(9), binary.BigEndian.Uint16(pkt[0:2])) // version + // count = template records (1) + data records (1) = 2 + require.Equal(t, uint16(2), binary.BigEndian.Uint16(pkt[2:4])) + require.Equal(t, uint32(0), binary.BigEndian.Uint32(pkt[12:16])) // package_sequence starts at 0 + + // First flowset after the 20-byte header is the template flowset (id 0). + fs := pkt[20:] + require.Equal(t, uint16(0), binary.BigEndian.Uint16(fs[0:2])) // flowset_id 0 = template + tmplLen := binary.BigEndian.Uint16(fs[2:4]) + require.Equal(t, templateID, binary.BigEndian.Uint16(fs[4:6])) // template_id + require.NotZero(t, binary.BigEndian.Uint16(fs[6:8])) // field count + + // Data flowset follows the template flowset. + data := fs[tmplLen:] + require.Equal(t, templateID, binary.BigEndian.Uint16(data[0:2])) // data flowset id = template id +} + +func TestV9TemplateRefreshByCount(t *testing.T) { + // refresh every 3 records; template should reappear on packet 1 and again + // once 3 records have been sent. + enc := NewV9Encoder(time.Unix(1699999000, 0), 0, 0) + enc.refreshRecords = 3 + enc.refreshInterval = time.Hour + now := time.Unix(1700000010, 0) + + p1 := enc.Encode(now, []embed.FlowRecord{testFlow(), testFlow()}) // 2 records, template included + require.True(t, hasTemplate(p1[0])) + p2 := enc.Encode(now, []embed.FlowRecord{testFlow()}) // crosses 3 → template again + require.True(t, hasTemplate(p2[0])) + p3 := enc.Encode(now, []embed.FlowRecord{testFlow()}) // no refresh due yet + require.False(t, hasTemplate(p3[0])) +} + +func hasTemplate(pkt []byte) bool { + return binary.BigEndian.Uint16(pkt[20:22]) == 0 +} diff --git a/flow/sflow/sflow.go b/flow/sflow/sflow.go new file mode 100644 index 00000000..05b25e8c --- /dev/null +++ b/flow/sflow/sflow.go @@ -0,0 +1,96 @@ +// Package sflow encodes blitz FlowRecords into sFlow v5 datagrams. Unlike the +// NetFlow/IPFIX family (which export flow summaries), sFlow exports sampled +// packets; here each FlowRecord becomes one flow sample carrying a sampled-IPv4 +// flow record (enterprise 0, format 3), the summary form that maps a 5-tuple +// without synthesizing a raw packet header. +package sflow + +import ( + "encoding/binary" + "net" + "sync/atomic" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/flow/wire" +) + +const ( + sampledIPv4Len = 32 // 8 uint32 fields + flowRecordLen = 8 + sampledIPv4Len + // sample body after the sample_length field: 8 uint32 header fields + + // one flow record. + sampleBodyLen = 8*4 + flowRecordLen + datagramHdr = 28 +) + +// Encoder encodes FlowRecords as sFlow v5 datagrams. Safe for concurrent Encode. +type Encoder struct { + agentIP net.IP + subAgentID uint32 + datagram atomic.Uint32 // datagram sequence + sampleSeq atomic.Uint32 // per-sample sequence +} + +// NewEncoder returns an sFlow encoder reporting agentIP as the exporter address. +func NewEncoder(agentIP net.IP, subAgentID uint32) *Encoder { + return &Encoder{agentIP: agentIP, subAgentID: subAgentID} +} + +// Encode renders flows into a single sFlow v5 datagram, one flow sample per +// record, stamped with uptime = now-boot. Returns nil for an empty batch. +func (e *Encoder) Encode(now, boot time.Time, flows []embed.FlowRecord) [][]byte { + if len(flows) == 0 { + return nil + } + buf := make([]byte, datagramHdr+len(flows)*(8+sampleBodyLen)) + + binary.BigEndian.PutUint32(buf[0:4], 5) // version + binary.BigEndian.PutUint32(buf[4:8], 1) // agent address type IPv4 + copy(buf[8:12], ip4(e.agentIP)) // agent address + binary.BigEndian.PutUint32(buf[12:16], e.subAgentID) // sub agent id + binary.BigEndian.PutUint32(buf[16:20], e.datagram.Add(1)-1) // datagram sequence + binary.BigEndian.PutUint32(buf[20:24], wire.U32(now.Sub(boot).Milliseconds())) // uptime + binary.BigEndian.PutUint32(buf[24:28], wire.U32(len(flows))) // num_samples + + off := datagramHdr + for _, f := range flows { + off += e.putSample(buf[off:], f) + } + return [][]byte{buf} +} + +func (e *Encoder) putSample(b []byte, f embed.FlowRecord) int { + binary.BigEndian.PutUint32(b[0:4], 1) // sample_data_format = flow sample + binary.BigEndian.PutUint32(b[4:8], wire.U32(sampleBodyLen)) // sample_length + binary.BigEndian.PutUint32(b[8:12], e.sampleSeq.Add(1)-1) // sample_sequence_number + binary.BigEndian.PutUint32(b[12:16], e.subAgentID) // source_id + binary.BigEndian.PutUint32(b[16:20], f.SamplingRate) // sampling_rate + binary.BigEndian.PutUint32(b[20:24], 0) // sample_pool + binary.BigEndian.PutUint32(b[24:28], 0) // drops + binary.BigEndian.PutUint32(b[28:32], f.InputIface) // input + binary.BigEndian.PutUint32(b[32:36], f.OutputIface) // output + binary.BigEndian.PutUint32(b[36:40], 1) // num_flow_records + + fr := b[40:] + binary.BigEndian.PutUint32(fr[0:4], 3) // flow_format = sampled ipv4 + binary.BigEndian.PutUint32(fr[4:8], wire.U32(sampledIPv4Len)) // flow_length + d := fr[8:] + binary.BigEndian.PutUint32(d[0:4], wire.U32(f.Bytes)) // length + binary.BigEndian.PutUint32(d[4:8], wire.U32(f.Protocol)) // protocol + copy(d[8:12], ip4(f.SrcIP)) + copy(d[12:16], ip4(f.DstIP)) + binary.BigEndian.PutUint32(d[16:20], wire.U32(f.SrcPort)) + binary.BigEndian.PutUint32(d[20:24], wire.U32(f.DstPort)) + binary.BigEndian.PutUint32(d[24:28], wire.U32(f.TCPFlags)) + binary.BigEndian.PutUint32(d[28:32], wire.U32(f.TOS)) + + return 8 + sampleBodyLen +} + +func ip4(ip net.IP) []byte { + if v4 := ip.To4(); v4 != nil { + return v4 + } + return []byte{0, 0, 0, 0} +} diff --git a/flow/sflow/sflow_test.go b/flow/sflow/sflow_test.go new file mode 100644 index 00000000..412cb62f --- /dev/null +++ b/flow/sflow/sflow_test.go @@ -0,0 +1,60 @@ +package sflow + +import ( + "encoding/binary" + "net" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/stretchr/testify/require" +) + +func testFlow() embed.FlowRecord { + return embed.FlowRecord{ + SrcIP: net.IPv4(10, 0, 0, 1), DstIP: net.IPv4(93, 184, 216, 34), + SrcPort: 54321, DstPort: 443, Protocol: 6, TCPFlags: 0x1b, + InputIface: 2, OutputIface: 3, Packets: 42, Bytes: 4200, + SamplingRate: 1024, + } +} + +func be(b []byte, o int) uint32 { return binary.BigEndian.Uint32(b[o : o+4]) } + +func TestSFlowDatagramHeader(t *testing.T) { + enc := NewEncoder(net.IPv4(192, 0, 2, 1), 0) + pkts := enc.Encode(time.Unix(1700000010, 0), time.Unix(1699999000, 0), []embed.FlowRecord{testFlow()}) + require.Len(t, pkts, 1) + d := pkts[0] + + require.Equal(t, uint32(5), be(d, 0)) // version + require.Equal(t, uint32(1), be(d, 4)) // agent addr type IPv4 + require.Equal(t, []byte{192, 0, 2, 1}, d[8:12]) + require.Equal(t, uint32(0), be(d, 12)) // sub_agent_id + require.Equal(t, uint32(0), be(d, 16)) // datagram seq starts 0 + require.Equal(t, uint32(1010*1000), be(d, 20)) // uptime ms + require.Equal(t, uint32(1), be(d, 24)) // num_samples +} + +func TestSFlowSampledIPv4(t *testing.T) { + enc := NewEncoder(net.IPv4(192, 0, 2, 1), 0) + d := enc.Encode(time.Unix(1700000010, 0), time.Unix(1699999000, 0), []embed.FlowRecord{testFlow()})[0] + + s := d[28:] // first sample begins after 28-byte datagram header + require.Equal(t, uint32(1), be(s, 0)) // sample_data_format = flow sample + // sample header: format,len,seq,source_id,rate,pool,drops,input,output,num_records + require.Equal(t, uint32(1024), be(s, 16)) // sampling_rate + require.Equal(t, uint32(2), be(s, 28)) // input + require.Equal(t, uint32(3), be(s, 32)) // output + require.Equal(t, uint32(1), be(s, 36)) // num_flow_records + + fr := s[40:] + require.Equal(t, uint32(3), be(fr, 0)) // flow_format = sampled ipv4 + d0 := fr[8:] // sampled_ipv4 data + require.Equal(t, uint32(4200), be(d0, 0)) // length (bytes) + require.Equal(t, uint32(6), be(d0, 4)) // protocol + require.Equal(t, []byte{10, 0, 0, 1}, d0[8:12]) // src ip + require.Equal(t, []byte{93, 184, 216, 34}, d0[12:16]) // dst ip + require.Equal(t, uint32(54321), be(d0, 16)) // src port + require.Equal(t, uint32(443), be(d0, 20)) // dst port +} diff --git a/flow/wire/wire.go b/flow/wire/wire.go new file mode 100644 index 00000000..e31a45db --- /dev/null +++ b/flow/wire/wire.go @@ -0,0 +1,32 @@ +// Package wire holds the fixed-width integer conversions the flow encoders use +// to pack values into NetFlow/IPFIX/sFlow wire fields. Each flow protocol +// defines its fields at a fixed width, so narrowing a wider in-memory value to +// that width is the protocol's defined behavior, not an accidental overflow — +// the gosec G115 suppression lives here, in one place, rather than on every +// call site. +package wire + +// integer is any integer type an encoder narrows from. +type integer interface { + ~int | ~int8 | ~int16 | ~int32 | ~int64 | ~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 +} + +// U32 narrows v to a 32-bit wire field. +func U32[T integer](v T) uint32 { + return uint32(v) // #nosec G115 -- fixed-width wire field; truncation is per protocol spec +} + +// U16 narrows v to a 16-bit wire field. +func U16[T integer](v T) uint16 { + return uint16(v) // #nosec G115 -- fixed-width wire field; truncation is per protocol spec +} + +// U8 narrows v to an 8-bit wire field. +func U8[T integer](v T) uint8 { + return uint8(v) // #nosec G115 -- fixed-width wire field; truncation is per protocol spec +} + +// U64 fits v into a 64-bit wire field (used for unsigned counters/timestamps). +func U64[T integer](v T) uint64 { + return uint64(v) // #nosec G115 -- fixed-width wire field; sign reinterpretation is per protocol spec +} diff --git a/generator/flow/flow.go b/generator/flow/flow.go new file mode 100644 index 00000000..81d9c394 --- /dev/null +++ b/generator/flow/flow.go @@ -0,0 +1,154 @@ +// Package flow generates network FlowRecords (the 5-tuple + counters) shaped +// by a canned traffic scenario. It is protocol-agnostic: the wire format +// (NetFlow v5/v9, IPFIX, sFlow) is chosen by the flow output that consumes +// these records, so one generator drives every exporter format. +package flow + +import ( + "context" + "fmt" + "math/rand" + "sync" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/generator" + "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/telemetry" + "go.uber.org/zap" +) + +const componentName = "flow" + +// Generator produces FlowRecords and pushes them to a FlowConsumer. +type Generator struct { + embed.ProducerMarker + + logger *zap.Logger + workers int + rate time.Duration + scenario Scenario + seed int64 + consumer embed.FlowConsumer + metrics *generator.Metrics + + wg sync.WaitGroup + stopCh chan struct{} + tracker *count.Tracker +} + +// New builds a flow generator. seed < 0 randomizes; 0+ is a deterministic seed. +func New(logger *zap.Logger, workers int, rate time.Duration, scenario Scenario, seed int64, consumer embed.FlowConsumer, tel embed.TelemetrySettings) (*Generator, error) { + if logger == nil { + return nil, fmt.Errorf("logger cannot be nil") + } + if workers < 1 { + return nil, fmt.Errorf("workers must be 1 or greater, got %d", workers) + } + if consumer == nil { + return nil, fmt.Errorf("consumer cannot be nil") + } + if !ValidScenario(scenario) { + return nil, fmt.Errorf("unknown flow scenario %q", scenario) + } + metrics, err := generator.NewMetrics(tel.MeterProvider) + if err != nil { + return nil, fmt.Errorf("build generator metrics: %w", err) + } + if scenario == "" { + scenario = ScenarioDefault + } + return &Generator{ + logger: logger, + workers: workers, + rate: rate, + scenario: scenario, + seed: seed, + consumer: consumer, + metrics: metrics, + stopCh: make(chan struct{}), + }, nil +} + +// Name returns the module identifier. +func (g *Generator) Name() string { return componentName } + +// SupportedTelemetry reports that this generator produces flow records. +func (g *Generator) SupportedTelemetry() []telemetry.Type { + return []telemetry.Type{telemetry.Flows} +} + +// SetCountTracker sets the finite generation count tracker. +func (g *Generator) SetCountTracker(t *count.Tracker) { g.tracker = t } + +// Start launches the worker goroutines. +func (g *Generator) Start(_ context.Context) error { + g.logger.Info("Starting flow generator", + zap.Int("workers", g.workers), zap.Duration("rate", g.rate), zap.String("scenario", string(g.scenario))) + g.metrics.BlitzGeneratorActiveWorkersGauge.Record(context.Background(), int64(g.workers), componentName) + for i := 0; i < g.workers; i++ { + g.wg.Add(1) + go g.worker(i) + } + return nil +} + +// Stop stops the generator and waits for workers to drain. +func (g *Generator) Stop(ctx context.Context) error { + g.logger.Info("Stopping flow generator") + g.metrics.BlitzGeneratorActiveWorkersGauge.Record(ctx, 0, componentName) + close(g.stopCh) + done := make(chan struct{}) + go func() { g.wg.Wait(); close(done) }() + select { + case <-done: + return nil + case <-ctx.Done(): + return fmt.Errorf("stop cancelled due to context cancellation: %w", ctx.Err()) + } +} + +func (g *Generator) worker(workerID int) { + defer g.wg.Done() + // Per-worker RNG: deterministic seed offsets by worker so a fixed seed + // reproduces the whole run while workers stay independent. seed < 0 + // randomizes from wall clock. + seed := g.seed + int64(workerID) + if g.seed < 0 { + seed = time.Now().UnixNano() + int64(workerID) + } + r := rand.New(rand.NewSource(seed)) // #nosec G404 -- synthetic load data, not security-sensitive + + ticker := time.NewTicker(g.rate) + defer ticker.Stop() + for { + select { + case <-g.stopCh: + return + case <-ticker.C: + if g.tracker != nil && !g.tracker.Acquire() { + select { + case <-g.stopCh: + return + case <-g.tracker.ResumeC(): + continue + } + } + if err := g.generateAndWrite(r); err != nil { + g.logger.Error("Failed to write flow", zap.Int("worker_id", workerID), zap.Error(err)) + } + } + } +} + +func (g *Generator) generateAndWrite(r *rand.Rand) error { + rec := generateFlow(r, g.scenario, time.Now()) + g.metrics.BlitzGeneratorEntriesCounter.Add(context.Background(), 1, componentName) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := g.consumer.ConsumeFlows(ctx, []embed.FlowRecord{rec}); err != nil { + g.metrics.BlitzGeneratorWriteErrorsCounter.Add(context.Background(), 1, componentName) + return err + } + return nil +} diff --git a/generator/flow/scenario.go b/generator/flow/scenario.go new file mode 100644 index 00000000..f7dac24c --- /dev/null +++ b/generator/flow/scenario.go @@ -0,0 +1,103 @@ +package flow + +import ( + "math/rand" + "net" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/flow/wire" + "github.com/observiq/blitz/internal/datagen" +) + +// Scenario names a canned traffic shape. Each preset controls the byte/packet +// size distribution and source/destination spread so the emitted flows +// resemble a recognizable environment. +type Scenario string + +const ( + // ScenarioDefault is a broad mix of flow sizes. + ScenarioDefault Scenario = "default" + // ScenarioWANEdge is few, large, bidirectional flows (low rate). + ScenarioWANEdge Scenario = "wan-edge" + // ScenarioDatacenter is many small flows (high rate). + ScenarioDatacenter Scenario = "datacenter" + // ScenarioDDOSTarget skews many sources onto a single destination. + ScenarioDDOSTarget Scenario = "ddos-target" +) + +// ValidScenario reports whether s is a known scenario. +func ValidScenario(s Scenario) bool { + switch s { + case ScenarioDefault, ScenarioWANEdge, ScenarioDatacenter, ScenarioDDOSTarget, "": + return true + default: + return false + } +} + +// shape holds a scenario's tunable ranges. Ranges are int64 so the +// rand.Int63n span math needs no unsigned conversion; the final counts are +// widened to the FlowRecord's uint64 fields. +type shape struct { + minBytes, maxBytes int64 + minPackets, maxPackets int64 + fixedDst bool +} + +func shapeFor(s Scenario) shape { + switch s { + case ScenarioWANEdge: + return shape{minBytes: 100_000, maxBytes: 10_000_000, minPackets: 100, maxPackets: 10_000} + case ScenarioDatacenter: + return shape{minBytes: 40, maxBytes: 1_500, minPackets: 1, maxPackets: 10} + case ScenarioDDOSTarget: + return shape{minBytes: 40, maxBytes: 200, minPackets: 1, maxPackets: 3, fixedDst: true} + default: + return shape{minBytes: 100, maxBytes: 1_000_000, minPackets: 1, maxPackets: 1_000} + } +} + +// ddosDst is the fixed victim for the ddos-target scenario. +var ddosDst = net.IPv4(203, 0, 113, 7) + +// generateFlow builds one FlowRecord shaped by the scenario, using r for all +// randomness so a fixed seed yields a deterministic stream. +func generateFlow(r *rand.Rand, s Scenario, now time.Time) embed.FlowRecord { + sh := shapeFor(s) + proto := []uint8{6, 17, 1}[r.Intn(3)] // TCP, UDP, ICMP + + dst := net.ParseIP(datagen.RandomPublicIPv4(r)) + dstPort := wire.U16(r.Intn(65535) + 1) + if sh.fixedDst { + dst = ddosDst + dstPort = 443 + } + + bytes := wire.U64(sh.minBytes + r.Int63n(sh.maxBytes-sh.minBytes+1)) + pkts := wire.U64(sh.minPackets + r.Int63n(sh.maxPackets-sh.minPackets+1)) + dur := time.Duration(r.Intn(10_000)) * time.Millisecond + + return embed.FlowRecord{ + SrcIP: net.ParseIP(datagen.RandomPublicIPv4(r)), + DstIP: dst, + SrcPort: wire.U16(r.Intn(65535) + 1), + DstPort: dstPort, + Protocol: proto, + TOS: 0, + TCPFlags: wire.U8(r.Intn(256)), + NextHop: net.ParseIP(datagen.RandomPrivateIPv4(r)), + InputIface: wire.U32(r.Intn(48) + 1), + OutputIface: wire.U32(r.Intn(48) + 1), + Packets: pkts, + Bytes: bytes, + StartTime: now.Add(-dur), + EndTime: now, + SrcAS: wire.U32(r.Intn(65000) + 1), + DstAS: wire.U32(r.Intn(65000) + 1), + SrcMask: wire.U8(r.Intn(9) + 24), + DstMask: wire.U8(r.Intn(9) + 24), + SamplingRate: 1024, + Metadata: embed.FlowRecordMetadata{Timestamp: now}, + } +} diff --git a/generator/flow/scenario_test.go b/generator/flow/scenario_test.go new file mode 100644 index 00000000..b511e8a3 --- /dev/null +++ b/generator/flow/scenario_test.go @@ -0,0 +1,36 @@ +package flow + +import ( + "math/rand" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func meanBytes(s Scenario, seed int64, n int) float64 { + r := rand.New(rand.NewSource(seed)) + var sum uint64 + for i := 0; i < n; i++ { + sum += generateFlow(r, s, time.Unix(1700000000, 0)).Bytes + } + return float64(sum) / float64(n) +} + +func TestWANEdgeFlowsLargerThanDatacenter(t *testing.T) { + require.Greater(t, meanBytes(ScenarioWANEdge, 1, 500), meanBytes(ScenarioDatacenter, 1, 500)) +} + +func TestDDOSTargetSharesDestination(t *testing.T) { + r := rand.New(rand.NewSource(1)) + first := generateFlow(r, ScenarioDDOSTarget, time.Unix(1700000000, 0)).DstIP + for i := 0; i < 100; i++ { + require.True(t, generateFlow(r, ScenarioDDOSTarget, time.Unix(1700000000, 0)).DstIP.Equal(first)) + } +} + +func TestDeterministicFromSeed(t *testing.T) { + a := generateFlow(rand.New(rand.NewSource(42)), ScenarioDefault, time.Unix(1700000000, 0)) + b := generateFlow(rand.New(rand.NewSource(42)), ScenarioDefault, time.Unix(1700000000, 0)) + require.Equal(t, a, b) +} diff --git a/go.mod b/go.mod index c6cec854..7193b202 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/goccy/go-json v0.10.6 github.com/google/uuid v1.6.0 github.com/hashicorp/golang-lru/v2 v2.0.7 + github.com/netsampler/goflow2/v2 v2.2.6 github.com/prometheus/client_golang v1.24.1 github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 diff --git a/go.sum b/go.sum index b697c1a6..ddedb39d 100644 --- a/go.sum +++ b/go.sum @@ -100,6 +100,8 @@ github.com/mgechev/revive v1.13.0 h1:yFbEVliCVKRXY8UgwEO7EOYNopvjb1BFbmYqm9hZjBM github.com/mgechev/revive v1.13.0/go.mod h1:efJfeBVCX2JUumNQ7dtOLDja+QKj9mYGgEZA7rt5u+0= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/netsampler/goflow2/v2 v2.2.6 h1:pm+UEykIYV+lGJzrunYLNehoQtHlipJTp3lFhBpUkj0= +github.com/netsampler/goflow2/v2 v2.2.6/go.mod h1:sKVfADpYP0NvkEeMRMh8/vz6T1yq4z1JZqO+N0+RGqk= github.com/onsi/ginkgo/v2 v2.32.1 h1:6tlvcDm/3sE8lGJbZ4+d4mO3RLy24/tQWOFzVSQNIfw= github.com/onsi/ginkgo/v2 v2.32.1/go.mod h1:+aXOY+vzZ5mu2iI2HpTZUPmM//oQfsNFX6gU9kNcA44= github.com/onsi/gomega v1.42.1 h1:iN1rCUX+44NZ1Dc97MPoeFYbFR0vh8zxoxMFwKdyZ6I= diff --git a/internal/config/generator.go b/internal/config/generator.go index 84e007cf..ee5f36b9 100644 --- a/internal/config/generator.go +++ b/internal/config/generator.go @@ -41,6 +41,9 @@ const ( // GeneratorTypeFIX represents the FIX (Financial Information // eXchange) protocol generator GeneratorTypeFIX GeneratorType = "fix" + // GeneratorTypeFlow represents the network flow (NetFlow/IPFIX/sFlow) + // generator + GeneratorTypeFlow GeneratorType = "flow" ) // Generator contains configuration for log generators @@ -79,6 +82,8 @@ type Generator struct { Wel WelGeneratorConfig `yaml:"wel,omitempty" mapstructure:"wel,omitempty"` // FIX contains FIX generator configuration FIX FIXGeneratorConfig `yaml:"fix,omitempty" mapstructure:"fix,omitempty"` + // Flow contains network flow generator configuration + Flow FlowGeneratorConfig `yaml:"flow,omitempty" mapstructure:"flow,omitempty"` } // Validate validates the generator configuration @@ -155,8 +160,12 @@ func (g *Generator) Validate() error { if err := g.FIX.Validate(); err != nil { return fmt.Errorf("fix generator validation failed: %w", err) } + case GeneratorTypeFlow: + if err := g.Flow.Validate(); err != nil { + return fmt.Errorf("flow generator validation failed: %w", err) + } default: - return fmt.Errorf("invalid generator type: %s, must be one of: nop, json, winevt, palo-alto, apache-common, apache-combined, apache-error, nginx, postgres, kubernetes, filegen, okta, hostmetrics, traces, wel, fix", g.Type) + return fmt.Errorf("invalid generator type: %s, must be one of: nop, json, winevt, palo-alto, apache-common, apache-combined, apache-error, nginx, postgres, kubernetes, filegen, okta, hostmetrics, traces, wel, fix, flow", g.Type) } return nil diff --git a/internal/config/generator_flow.go b/internal/config/generator_flow.go new file mode 100644 index 00000000..22014aea --- /dev/null +++ b/internal/config/generator_flow.go @@ -0,0 +1,47 @@ +package config + +import ( + "fmt" + "time" +) + +// Default flow generator configuration values. +const ( + // DefaultFlowScenario is the traffic-shape preset used when none is set. + DefaultFlowScenario = "default" + // DefaultFlowSeed randomizes each run (negative = randomize per the + // datagen seed contract). + DefaultFlowSeed = -1 +) + +// validFlowScenarios is the set of accepted traffic-shape presets. +var validFlowScenarios = map[string]bool{ + "": true, "default": true, "wan-edge": true, "datacenter": true, "ddos-target": true, +} + +// FlowGeneratorConfig contains configuration for the network flow generator. +type FlowGeneratorConfig struct { + // Workers is the number of worker goroutines. + Workers int `yaml:"workers,omitempty" mapstructure:"workers,omitempty"` + // Rate is the generation interval per worker. + Rate time.Duration `yaml:"rate,omitempty" mapstructure:"rate,omitempty"` + // Scenario is the traffic-shape preset: default, wan-edge, datacenter, + // ddos-target. + Scenario string `yaml:"scenario,omitempty" mapstructure:"scenario,omitempty"` + // Seed is the RNG seed; negative randomizes, 0+ is deterministic. + Seed int64 `yaml:"seed,omitempty" mapstructure:"seed,omitempty"` +} + +// Validate validates the flow generator configuration. +func (c *FlowGeneratorConfig) Validate() error { + if c.Workers < 1 { + return fmt.Errorf("flow generator workers must be 1 or greater, got %d", c.Workers) + } + if c.Rate <= 0 { + return fmt.Errorf("flow generator rate must be positive, got %v", c.Rate) + } + if !validFlowScenarios[c.Scenario] { + return fmt.Errorf("flow generator scenario must be one of: default, wan-edge, datacenter, ddos-target; got %q", c.Scenario) + } + return nil +} diff --git a/internal/config/output.go b/internal/config/output.go index 78b3e294..342dfb35 100644 --- a/internal/config/output.go +++ b/internal/config/output.go @@ -24,6 +24,8 @@ const ( OutputTypeFile OutputType = "file" // OutputTypeHEC represents Splunk HEC output OutputTypeHEC OutputType = "hec" + // OutputTypeFlow represents the network flow (NetFlow/IPFIX/sFlow) output + OutputTypeFlow OutputType = "flow" ) // Output contains configuration for output destinations @@ -44,6 +46,8 @@ type Output struct { HEC HECOutputConfig `yaml:"hec,omitempty" mapstructure:"hec,omitempty"` // Stdout contains stdout output configuration Stdout StdoutOutputConfig `yaml:"stdout,omitempty" mapstructure:"stdout,omitempty"` + // Flow contains network flow output configuration + Flow FlowOutputConfig `yaml:"flow,omitempty" mapstructure:"flow,omitempty"` } // Validate validates the output configuration @@ -84,8 +88,12 @@ func (o *Output) Validate() error { if err := o.HEC.Validate(); err != nil { return fmt.Errorf("HEC output validation failed: %w", err) } + case OutputTypeFlow: + if err := o.Flow.Validate(); err != nil { + return fmt.Errorf("flow output validation failed: %w", err) + } default: - return fmt.Errorf("invalid output type: %s, must be one of: nop, stdout, tcp, udp, syslog, otlp-grpc, file, hec", o.Type) + return fmt.Errorf("invalid output type: %s, must be one of: nop, stdout, tcp, udp, syslog, otlp-grpc, file, hec, flow", o.Type) } return nil diff --git a/internal/config/output_flow.go b/internal/config/output_flow.go new file mode 100644 index 00000000..f384208b --- /dev/null +++ b/internal/config/output_flow.go @@ -0,0 +1,92 @@ +package config + +import "fmt" + +// Default flow output configuration values. +const ( + // DefaultFlowOutputHost is the collector host. + DefaultFlowOutputHost = "127.0.0.1" + // DefaultFlowOutputPort is the common NetFlow/IPFIX collector port. + DefaultFlowOutputPort = 2055 + // DefaultFlowProtocol is the wire format used when none is set. + DefaultFlowProtocol = "netflow-v9" +) + +// validFlowProtocols is the set of accepted wire formats. +var validFlowProtocols = map[string]bool{ + "": true, "netflow-v5": true, "netflow-v9": true, "ipfix": true, "sflow": true, +} + +// flowVendorFormats maps a vendor flavor to every wire format that vendor's +// exporter actually supports in the real world. A vendor is mutually exclusive +// with the others by construction (one field), and the configured protocol must +// be one the vendor supports. +// +// Sources (per row): Juniper jFlow (v5/v9/IPFIX — Juniper "Flow Monitoring" +// docs); Huawei NetStream (v5/v9/IPFIX — Huawei NetStream config guide); Nokia +// cflowd (v5/v8/v9/IPFIX — Nokia SR OS cflowd; v8 is aggregation-only and +// excluded here, see note); Citrix AppFlow (IPFIX only — Citrix AppFlow is an +// IPFIX application); Redback rFlow (v5/v9 — Redback/Ericsson SmartEdge rFlow). +// +// How the vendor distinction rides each format: IPFIX carries a decoder-readable +// enterprise IE (the vendor's IANA PEN); NetFlow v9 has no PEN mechanism, so the +// signature is a proprietary field type in the template; NetFlow v5 is a fixed +// layout with no vendor mechanism at all, so a v5 export is byte-identical +// across vendors (the pairing is allowed for fidelity, but carries no +// vendor-distinct wire element — this is the real protocol's behavior). +var flowVendorFormats = map[string][]string{ + "jflow": {"netflow-v5", "netflow-v9", "ipfix"}, // Juniper, PEN 2636 + "netstream": {"netflow-v5", "netflow-v9", "ipfix"}, // Huawei, PEN 2011 + "cflowd": {"netflow-v5", "netflow-v9", "ipfix"}, // Nokia/Alcatel-Lucent, PEN 6527 + "appflow": {"ipfix"}, // Citrix, PEN 5951 + "rflow": {"netflow-v5", "netflow-v9"}, // Redback/Ericsson +} + +// FlowOutputConfig contains configuration for the flow output. +type FlowOutputConfig struct { + // Host is the collector host. + Host string `yaml:"host,omitempty" mapstructure:"host,omitempty"` + // Port is the collector UDP port. + Port int `yaml:"port,omitempty" mapstructure:"port,omitempty"` + // Protocol is the wire format: netflow-v5, netflow-v9, ipfix, sflow. + Protocol string `yaml:"protocol,omitempty" mapstructure:"protocol,omitempty"` + // Vendor is an optional vendor flavor (jflow, netstream, cflow, appflow, + // rflow); it must be compatible with Protocol. Empty means standard. + Vendor string `yaml:"vendor,omitempty" mapstructure:"vendor,omitempty"` + // AgentIP is the exporter address reported by sFlow (optional). + AgentIP string `yaml:"agentIP,omitempty" mapstructure:"agentIP,omitempty"` +} + +// Validate validates the flow output configuration, including the +// vendor/protocol single-flavor compatibility rule. +func (c *FlowOutputConfig) Validate() error { + if c.Port < 0 || c.Port > 65535 { + return fmt.Errorf("flow output port must be between 0 and 65535, got %d", c.Port) + } + if !validFlowProtocols[c.Protocol] { + return fmt.Errorf("flow output protocol must be one of: netflow-v5, netflow-v9, ipfix, sflow; got %q", c.Protocol) + } + if c.Vendor != "" { + formats, ok := flowVendorFormats[c.Vendor] + if !ok { + return fmt.Errorf("flow output vendor must be one of: appflow, jflow, cflowd, netstream, rflow; got %q", c.Vendor) + } + proto := c.Protocol + if proto == "" { + proto = DefaultFlowProtocol + } + if !contains(formats, proto) { + return fmt.Errorf("flow output vendor %q does not support protocol %q; supported: %v", c.Vendor, proto, formats) + } + } + return nil +} + +func contains(s []string, v string) bool { + for _, x := range s { + if x == v { + return true + } + } + return false +} diff --git a/internal/config/output_flow_test.go b/internal/config/output_flow_test.go new file mode 100644 index 00000000..2c7cb479 --- /dev/null +++ b/internal/config/output_flow_test.go @@ -0,0 +1,38 @@ +package config + +import "testing" + +func TestFlowOutputVendorFormatMatrix(t *testing.T) { + tests := []struct { + name string + protocol string + vendor string + wantErr bool + }{ + {"no vendor ipfix", "ipfix", "", false}, + {"appflow ipfix ok", "ipfix", "appflow", false}, + {"appflow v9 rejected", "netflow-v9", "appflow", true}, + {"appflow v5 rejected", "netflow-v5", "appflow", true}, + {"jflow v5 ok", "netflow-v5", "jflow", false}, + {"jflow v9 ok", "netflow-v9", "jflow", false}, + {"jflow ipfix ok", "ipfix", "jflow", false}, + {"netstream v5 ok", "netflow-v5", "netstream", false}, + {"netstream ipfix ok", "ipfix", "netstream", false}, + {"rflow v9 ok", "netflow-v9", "rflow", false}, + {"rflow ipfix rejected", "ipfix", "rflow", true}, + {"cflowd v9 ok", "netflow-v9", "cflowd", false}, + {"unknown vendor rejected", "ipfix", "bogus", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := FlowOutputConfig{Host: "127.0.0.1", Port: 2055, Protocol: tt.protocol, Vendor: tt.vendor} + err := c.Validate() + if tt.wantErr && err == nil { + t.Fatalf("expected error for protocol=%q vendor=%q, got nil", tt.protocol, tt.vendor) + } + if !tt.wantErr && err != nil { + t.Fatalf("unexpected error for protocol=%q vendor=%q: %v", tt.protocol, tt.vendor, err) + } + }) + } +} diff --git a/internal/config/override.go b/internal/config/override.go index e0594187..77255f55 100644 --- a/internal/config/override.go +++ b/internal/config/override.go @@ -284,6 +284,10 @@ func DefaultOverrides() []*Override { NewOverride("generator.hostmetrics.scrapers", "scrapers to enable (empty = all). One or more of: cpu|memory|disk|network|filesystem|load|paging|processes", []string{}), NewOverride("generator.traces.workers", "number of traces generator workers", 1), NewOverride("generator.traces.rate", "rate at which traces are generated per worker", 1*time.Second), + NewOverride("generator.flow.workers", "number of flow generator workers", 1), + NewOverride("generator.flow.rate", "rate at which flow records are generated per worker", 1*time.Second), + NewOverride("generator.flow.scenario", "flow traffic-shape preset. One of: default|wan-edge|datacenter|ddos-target", DefaultFlowScenario), + NewOverride("generator.flow.seed", "flow generator RNG seed (negative randomizes)", DefaultFlowSeed), NewOverride("generator.wel.workers", "number of WEL generator workers", 1), NewOverride("generator.wel.rate", "rate at which WEL events are generated per worker", 1*time.Second), NewOverride("generator.wel.channels", "WEL channels to generate events for (empty = all)", []string{}), @@ -338,6 +342,11 @@ func DefaultOverrides() []*Override { NewOverride("output.hec.source", "HEC event source metadata", DefaultHECSource), NewOverride("output.hec.sourceType", "HEC event sourcetype metadata", DefaultHECSourceType), NewOverride("output.hec.index", "HEC target index (empty = token default)", ""), + NewOverride("output.flow.host", "flow collector host", DefaultFlowOutputHost), + NewOverride("output.flow.port", "flow collector UDP port", DefaultFlowOutputPort), + NewOverride("output.flow.protocol", "flow wire format. One of: netflow-v5|netflow-v9|ipfix|sflow", DefaultFlowProtocol), + NewOverride("output.flow.vendor", "optional flow vendor flavor. One of: jflow|netstream|cflow|appflow|rflow", ""), + NewOverride("output.flow.agentIP", "sFlow exporter agent IP (sflow only)", ""), NewOverride("telemetry.traces.otlpEndpoint", "OTLP gRPC endpoint (host:port) for exporting blitz's own spans (empty = disabled)", ""), NewOverride("telemetry.traces.insecure", "send blitz's own spans over plaintext gRPC (no TLS)", false), NewOverride("telemetry.traces.perBatchSpans", "enable higher-volume per-emit-cycle spans (off by default)", false), diff --git a/internal/config/override_test.go b/internal/config/override_test.go index a81ae05c..faf3022e 100644 --- a/internal/config/override_test.go +++ b/internal/config/override_test.go @@ -60,6 +60,10 @@ func getTestOverrideFlagsArgs() []string { "--generator-hostmetrics-scrapers", "cpu,memory", "--generator-traces-workers", "2", "--generator-traces-rate", "500ms", + "--generator-flow-workers", "3", + "--generator-flow-rate", "250ms", + "--generator-flow-scenario", "wan-edge", + "--generator-flow-seed", "7", "--generator-wel-workers", "3", "--generator-wel-rate", "500ms", "--generator-wel-channels", "Security,System", @@ -132,6 +136,11 @@ func getTestOverrideFlagsArgs() []string { "--output-hec-source", "myapp", "--output-hec-sourcetype", "mylog", "--output-hec-index", "main", + "--output-flow-host", "10.0.0.9", + "--output-flow-port", "4739", + "--output-flow-protocol", "ipfix", + "--output-flow-vendor", "appflow", + "--output-flow-agentip", "10.0.0.1", "--output-hec-enable-tls", "true", "--output-hec-tls-cert", "/path/to/hec_cert.pem", "--output-hec-tls-key", "/path/to/hec_key.pem", @@ -197,6 +206,10 @@ func getTestOverrideEnvs() map[string]string { "BLITZ_GENERATOR_HOSTMETRICS_SCRAPERS": "disk,network", "BLITZ_GENERATOR_TRACES_WORKERS": "5", "BLITZ_GENERATOR_TRACES_RATE": "2s", + "BLITZ_GENERATOR_FLOW_WORKERS": "5", + "BLITZ_GENERATOR_FLOW_RATE": "750ms", + "BLITZ_GENERATOR_FLOW_SCENARIO": "datacenter", + "BLITZ_GENERATOR_FLOW_SEED": "11", "BLITZ_GENERATOR_WEL_WORKERS": "3", "BLITZ_GENERATOR_WEL_RATE": "500ms", "BLITZ_GENERATOR_WEL_CHANNELS": "Security,System", @@ -269,6 +282,11 @@ func getTestOverrideEnvs() map[string]string { "BLITZ_OUTPUT_HEC_SOURCE": "envapp", "BLITZ_OUTPUT_HEC_SOURCETYPE": "envlog", "BLITZ_OUTPUT_HEC_INDEX": "dev", + "BLITZ_OUTPUT_FLOW_HOST": "10.0.0.8", + "BLITZ_OUTPUT_FLOW_PORT": "6343", + "BLITZ_OUTPUT_FLOW_PROTOCOL": "netflow-v9", + "BLITZ_OUTPUT_FLOW_VENDOR": "rflow", + "BLITZ_OUTPUT_FLOW_AGENTIP": "10.0.0.2", "BLITZ_OUTPUT_HEC_ENABLE_TLS": "false", "BLITZ_OUTPUT_HEC_TLS_CERT": "/env/hec_cert.pem", "BLITZ_OUTPUT_HEC_TLS_KEY": "/env/hec_key.pem", @@ -378,6 +396,12 @@ func TestOverrideDefaults(t *testing.T) { Workers: 1, Rate: 1 * time.Second, }, + Flow: FlowGeneratorConfig{ + Workers: 1, + Rate: 1 * time.Second, + Scenario: DefaultFlowScenario, + Seed: DefaultFlowSeed, + }, Wel: WelGeneratorConfig{ Workers: 1, Rate: 1 * time.Second, @@ -467,6 +491,11 @@ func TestOverrideDefaults(t *testing.T) { Stdout: StdoutOutputConfig{ FlushInterval: DefaultStdoutFlushInterval, }, + Flow: FlowOutputConfig{ + Host: DefaultFlowOutputHost, + Port: DefaultFlowOutputPort, + Protocol: DefaultFlowProtocol, + }, }, Metrics: Metrics{ Port: DefaultMetricsPort, @@ -578,6 +607,12 @@ func TestOverrideFlags(t *testing.T) { Workers: 2, Rate: 500 * time.Millisecond, }, + Flow: FlowGeneratorConfig{ + Workers: 3, + Rate: 250 * time.Millisecond, + Scenario: "wan-edge", + Seed: 7, + }, Wel: WelGeneratorConfig{ Workers: 3, Rate: 500 * time.Millisecond, @@ -683,6 +718,13 @@ func TestOverrideFlags(t *testing.T) { Stdout: StdoutOutputConfig{ FlushInterval: 50 * time.Millisecond, }, + Flow: FlowOutputConfig{ + Host: "10.0.0.9", + Port: 4739, + Protocol: "ipfix", + Vendor: "appflow", + AgentIP: "10.0.0.1", + }, }, Metrics: Metrics{ Port: 8080, @@ -799,6 +841,12 @@ func TestOverrideEnvs(t *testing.T) { Workers: 5, Rate: 2 * time.Second, }, + Flow: FlowGeneratorConfig{ + Workers: 5, + Rate: 750 * time.Millisecond, + Scenario: "datacenter", + Seed: 11, + }, Wel: WelGeneratorConfig{ Workers: 3, Rate: 500 * time.Millisecond, @@ -903,6 +951,13 @@ func TestOverrideEnvs(t *testing.T) { Stdout: StdoutOutputConfig{ FlushInterval: 75 * time.Millisecond, }, + Flow: FlowOutputConfig{ + Host: "10.0.0.8", + Port: 6343, + Protocol: "netflow-v9", + Vendor: "rflow", + AgentIP: "10.0.0.2", + }, }, Metrics: Metrics{ Port: 9100, diff --git a/internal/dispatch/embed.go b/internal/dispatch/embed.go index d6844f5b..4e13271e 100644 --- a/internal/dispatch/embed.go +++ b/internal/dispatch/embed.go @@ -16,6 +16,7 @@ import ( "github.com/observiq/blitz/generator/filegen" fixgen "github.com/observiq/blitz/generator/fix" "github.com/observiq/blitz/generator/fix/catalog" + flowgen "github.com/observiq/blitz/generator/flow" "github.com/observiq/blitz/generator/hostmetrics" jsongen "github.com/observiq/blitz/generator/json" "github.com/observiq/blitz/generator/kubernetes" @@ -42,6 +43,7 @@ type EmbedConsumers struct { LogConsumer embed.LogConsumer MetricConsumer embed.MetricConsumer TraceConsumer embed.TraceConsumer + FlowConsumer embed.FlowConsumer } func (c EmbedConsumers) requireLog(typ config.GeneratorType) error { @@ -65,6 +67,13 @@ func (c EmbedConsumers) requireTrace(typ config.GeneratorType) error { return nil } +func (c EmbedConsumers) requireFlow(typ config.GeneratorType) error { + if c.FlowConsumer == nil { + return fmt.Errorf("generator type %q requires EmbedConsumers.FlowConsumer", typ) + } + return nil +} + // ForEmbed constructs an embed.ProducerModule for the given generator // config wired to the relevant consumer in `consumers`. fileGenLibrary // is optional and only consulted by the filegen generator: pass @@ -172,6 +181,11 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum } mod, err := newFIX(logger, genCfg.FIX, consumers.LogConsumer, tel) return applyHostIdentity(mod, err, env, genCfg.Type) + case config.GeneratorTypeFlow: + if err := consumers.requireFlow(genCfg.Type); err != nil { + return nil, err + } + return flowgen.New(logger, genCfg.Flow.Workers, genCfg.Flow.Rate, flowgen.Scenario(genCfg.Flow.Scenario), genCfg.Flow.Seed, consumers.FlowConsumer, tel) case config.GeneratorTypeHostMetrics: if err := consumers.requireMetric(genCfg.Type); err != nil { return nil, err diff --git a/output/flow/flow.go b/output/flow/flow.go new file mode 100644 index 00000000..29b748f8 --- /dev/null +++ b/output/flow/flow.go @@ -0,0 +1,180 @@ +// Package flow is the flow output: it consumes FlowRecords, encodes them into +// the configured wire format (NetFlow v5/v9, IPFIX, or sFlow), and pushes the +// packets over UDP — the transport every flow collector expects. +package flow + +import ( + "context" + "fmt" + "net" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/flow/ipfix" + "github.com/observiq/blitz/flow/netflow" + "github.com/observiq/blitz/flow/sflow" + "github.com/observiq/blitz/output" + "github.com/observiq/blitz/telemetry" + "go.uber.org/zap" +) + +// Protocol is a flow wire format. +type Protocol string + +const ( + // ProtocolNetFlowV5 is fixed-layout NetFlow v5. + ProtocolNetFlowV5 Protocol = "netflow-v5" + // ProtocolNetFlowV9 is template-based NetFlow v9. + ProtocolNetFlowV9 Protocol = "netflow-v9" + // ProtocolIPFIX is IPFIX (RFC 7011). + ProtocolIPFIX Protocol = "ipfix" + // ProtocolSFlow is sFlow v5. + ProtocolSFlow Protocol = "sflow" +) + +// encoder is the format-agnostic surface the output drives. Each wire encoder +// packs a batch of records into one or more UDP payloads. +type encoder interface { + encode(now time.Time, flows []embed.FlowRecord) [][]byte +} + +// Output implements output.Output (metrics-adjacent: flows only) and +// embed.FlowConsumer. Write returns ErrUnsupportedTelemetryType because flows +// are delivered via ConsumeFlows, not the log Write path. +type Output struct { + logger *zap.Logger + conn *net.UDPConn + enc encoder + metrics *output.Metrics +} + +// vendorSpec is how a vendor flavor diverges on the wire, per format. A vendor +// may support several formats (see config.flowVendorFormats); the encoder picks +// the signature matching the configured protocol: +// - IPFIX → an enterprise IE carrying the vendor's IANA PEN (pen/ieid). +// - NetFlow v9 → a proprietary field type in the template (v9Field). +// - NetFlow v5 → no signature: v5 is a fixed layout with no vendor mechanism, +// so a v5 export is byte-identical across vendors (real protocol behavior). +// +// pen/v9Field are zero for a format the vendor does not carry a mark in. +type vendorSpec struct { + pen uint32 // IPFIX enterprise PEN + ieid uint16 // IPFIX enterprise element id + v9Field uint16 // NetFlow v9 proprietary field type +} + +// vendorSpecs holds each vendor's per-format signatures, grounded in the +// vendor's real IANA Private Enterprise Number. AppFlow is IPFIX-only, so it has +// no v9 field; the v5-only path carries no mark for any vendor. +var vendorSpecs = map[string]vendorSpec{ + "jflow": {pen: 2636, ieid: 137, v9Field: 0x9003}, // Juniper + "netstream": {pen: 2011, ieid: 140, v9Field: 0x9001}, // Huawei + "cflowd": {pen: 6527, ieid: 91, v9Field: 0x9004}, // Nokia/Alcatel-Lucent + "appflow": {pen: 5951, ieid: 130}, // Citrix (IPFIX only) + "rflow": {v9Field: 0x9002}, // Redback (v5/v9; no PEN) +} + +// ipfixVendor returns the IPFIX enterprise IE for a vendor, or nil when the +// vendor carries no IPFIX PEN. +func (s vendorSpec) ipfixVendor() *ipfix.Vendor { + if s.pen == 0 { + return nil + } + return &ipfix.Vendor{PEN: s.pen, IEID: s.ieid} +} + +// New dials the collector at host:port over UDP and builds the encoder for +// protocol. vendor (optional) selects a vendor-distinct wire encoding path. +// agentIP is the exporter address reported by sFlow (ignored by the +// NetFlow/IPFIX formats). +func New(logger *zap.Logger, host, port string, protocol Protocol, vendor, agentIP string, tel embed.TelemetrySettings) (*Output, error) { + if logger == nil { + return nil, fmt.Errorf("logger cannot be nil") + } + addr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(host, port)) + if err != nil { + return nil, fmt.Errorf("flow output: resolve %s:%s: %w", host, port, err) + } + conn, err := net.DialUDP("udp", nil, addr) + if err != nil { + return nil, fmt.Errorf("flow output: dial %s: %w", addr, err) + } + m, err := output.NewMetrics(tel.MeterProvider) + if err != nil { + _ = conn.Close() + return nil, fmt.Errorf("flow output: build metrics: %w", err) + } + + spec := vendorSpecs[vendor] // zero value (no signature) when vendor == "" + boot := time.Now() + var enc encoder + switch protocol { + case ProtocolNetFlowV5: + enc = v5enc{netflow.NewV5Encoder(boot)} + case ProtocolNetFlowV9: + enc = v9enc{netflow.NewV9Encoder(boot, 0, spec.v9Field)} + case ProtocolIPFIX: + enc = ipfixenc{ipfix.NewEncoder(0, spec.ipfixVendor())} + case ProtocolSFlow: + ip := net.ParseIP(agentIP) + if ip == nil { + ip = net.IPv4(127, 0, 0, 1) + } + enc = sflowenc{e: sflow.NewEncoder(ip, 0), boot: boot} + default: + _ = conn.Close() + return nil, fmt.Errorf("flow output: unknown protocol %q", protocol) + } + + return &Output{logger: logger, conn: conn, enc: enc, metrics: m}, nil +} + +// ConsumeFlows encodes the batch and sends each packet over UDP. +func (o *Output) ConsumeFlows(ctx context.Context, records []embed.FlowRecord) error { + for _, pkt := range o.enc.encode(time.Now(), records) { + if _, err := o.conn.Write(pkt); err != nil { + return fmt.Errorf("flow output: send: %w", err) + } + } + o.metrics.BlitzOutputEntriesReceivedCounter.Add(ctx, int64(len(records)), outputType, "flows") + return nil +} + +// Write reports that logs are unsupported: this output is flows-only. +func (o *Output) Write(_ context.Context, _ output.LogRecord) error { + return output.ErrUnsupportedTelemetryType +} + +// SupportedTelemetry reports that only flows are consumed. +func (o *Output) SupportedTelemetry() []telemetry.Type { + return []telemetry.Type{telemetry.Flows} +} + +// Stop closes the UDP socket. +func (o *Output) Stop(_ context.Context) error { + return o.conn.Close() +} + +const outputType = "flow" + +// encoder adapters — the wire encoders share a shape but not an interface. +type v5enc struct{ e *netflow.V5Encoder } + +func (a v5enc) encode(now time.Time, f []embed.FlowRecord) [][]byte { return a.e.Encode(now, f) } + +type v9enc struct{ e *netflow.V9Encoder } + +func (a v9enc) encode(now time.Time, f []embed.FlowRecord) [][]byte { return a.e.Encode(now, f) } + +type ipfixenc struct{ e *ipfix.Encoder } + +func (a ipfixenc) encode(now time.Time, f []embed.FlowRecord) [][]byte { return a.e.Encode(now, f) } + +type sflowenc struct { + e *sflow.Encoder + boot time.Time +} + +func (a sflowenc) encode(now time.Time, f []embed.FlowRecord) [][]byte { + return a.e.Encode(now, a.boot, f) +} diff --git a/output/flow/flow_test.go b/output/flow/flow_test.go new file mode 100644 index 00000000..3858bd3f --- /dev/null +++ b/output/flow/flow_test.go @@ -0,0 +1,45 @@ +package flow + +import ( + "context" + "encoding/binary" + "net" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/output" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func TestConsumeFlowsSendsEncodedPacket(t *testing.T) { + srv, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}) + require.NoError(t, err) + defer srv.Close() + _, portStr, _ := net.SplitHostPort(srv.LocalAddr().String()) + + o, err := New(zap.NewNop(), "127.0.0.1", portStr, ProtocolNetFlowV5, "", "", embed.TelemetrySettings{}) + require.NoError(t, err) + defer o.Stop(context.Background()) + + rec := embed.FlowRecord{ + SrcIP: net.IPv4(10, 0, 0, 1), DstIP: net.IPv4(8, 8, 8, 8), + SrcPort: 1234, DstPort: 53, Protocol: 17, Packets: 1, Bytes: 60, + } + require.NoError(t, o.ConsumeFlows(context.Background(), []embed.FlowRecord{rec})) + + buf := make([]byte, 2048) + require.NoError(t, srv.SetReadDeadline(time.Now().Add(2*time.Second))) + n, _, err := srv.ReadFromUDP(buf) + require.NoError(t, err) + require.Equal(t, uint16(5), binary.BigEndian.Uint16(buf[0:2])) // NetFlow v5 version + require.Equal(t, 24+48, n) // header + one record +} + +func TestWriteUnsupported(t *testing.T) { + o, err := New(zap.NewNop(), "127.0.0.1", "9995", ProtocolIPFIX, "", "", embed.TelemetrySettings{}) + require.NoError(t, err) + defer o.Stop(context.Background()) + require.ErrorIs(t, o.Write(context.Background(), output.LogRecord{}), output.ErrUnsupportedTelemetryType) +} diff --git a/output/otlp_grpc/flows.go b/output/otlp_grpc/flows.go new file mode 100644 index 00000000..aab27525 --- /dev/null +++ b/output/otlp_grpc/flows.go @@ -0,0 +1,31 @@ +package otlpgrpc + +import ( + "context" + "fmt" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/embed/otelpdata" + "github.com/observiq/blitz/output" +) + +// ConsumeFlows sends flow records to the OTLP gRPC output as OTLP logs, using +// the same network-flow semantic conventions the collector's netflowreceiver +// emits (via otelpdata.FlowToLogRecord). This makes the OTLP path a first-class +// destination for the flow signal, alongside logs/metrics/traces. +func (o *OTLPGrpc) ConsumeFlows(ctx context.Context, records []embed.FlowRecord) error { + for i := range records { + rec := otelpdata.FlowToLogRecord(records[i]) + rec.ObservedTimeUnixNano = output.TimeToUnixNanoUint64(time.Now()) + select { + case o.dataChan <- rec: + o.metrics.BlitzOutputEntriesReceivedCounter.Add(ctx, 1, outputType, "flows") + case <-ctx.Done(): + return fmt.Errorf("context cancelled while waiting to write flow: %w", ctx.Err()) + case <-o.ctx.Done(): + return fmt.Errorf("OTLP gRPC output is shutting down") + } + } + return nil +} diff --git a/package/completions/blitz.bash b/package/completions/blitz.bash index b4ceb56e..c4192888 100644 --- a/package/completions/blitz.bash +++ b/package/completions/blitz.bash @@ -399,6 +399,14 @@ _blitz_help() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -512,6 +520,16 @@ _blitz_help() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -685,6 +703,14 @@ _blitz_library_diff() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -798,6 +824,16 @@ _blitz_library_diff() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -972,6 +1008,14 @@ _blitz_library_extract() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -1085,6 +1129,16 @@ _blitz_library_extract() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -1257,6 +1311,14 @@ _blitz_library_ls() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -1370,6 +1432,16 @@ _blitz_library_ls() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -1542,6 +1614,14 @@ _blitz_library_path() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -1655,6 +1735,16 @@ _blitz_library_path() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -1827,6 +1917,14 @@ _blitz_library_search() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -1940,6 +2038,16 @@ _blitz_library_search() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -2112,6 +2220,14 @@ _blitz_library_show() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -2225,6 +2341,16 @@ _blitz_library_show() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -2403,6 +2529,14 @@ _blitz_library() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -2516,6 +2650,16 @@ _blitz_library() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -2688,6 +2832,14 @@ _blitz_version() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -2801,6 +2953,16 @@ _blitz_version() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") @@ -2976,6 +3138,14 @@ _blitz_root_command() two_word_flags+=("--generator-filegen-source") flags+=("--generator-filegen-workers=") two_word_flags+=("--generator-filegen-workers") + flags+=("--generator-flow-rate=") + two_word_flags+=("--generator-flow-rate") + flags+=("--generator-flow-scenario=") + two_word_flags+=("--generator-flow-scenario") + flags+=("--generator-flow-seed=") + two_word_flags+=("--generator-flow-seed") + flags+=("--generator-flow-workers=") + two_word_flags+=("--generator-flow-workers") flags+=("--generator-hostmetrics-hostname=") two_word_flags+=("--generator-hostmetrics-hostname") flags+=("--generator-hostmetrics-os=") @@ -3089,6 +3259,16 @@ _blitz_root_command() two_word_flags+=("--output-file-rotation-maxsizemb") flags+=("--output-file-workers=") two_word_flags+=("--output-file-workers") + flags+=("--output-flow-agentip=") + two_word_flags+=("--output-flow-agentip") + flags+=("--output-flow-host=") + two_word_flags+=("--output-flow-host") + flags+=("--output-flow-port=") + two_word_flags+=("--output-flow-port") + flags+=("--output-flow-protocol=") + two_word_flags+=("--output-flow-protocol") + flags+=("--output-flow-vendor=") + two_word_flags+=("--output-flow-vendor") flags+=("--output-hec-ackpollinterval=") two_word_flags+=("--output-hec-ackpollinterval") flags+=("--output-hec-acktimeout=") diff --git a/telemetry/telemetry.go b/telemetry/telemetry.go index 4b55f1b9..a8371932 100644 --- a/telemetry/telemetry.go +++ b/telemetry/telemetry.go @@ -13,12 +13,14 @@ const ( Metrics Type = "metrics" // Traces represents trace telemetry. Traces Type = "traces" + // Flows represents network flow telemetry (NetFlow/IPFIX/sFlow). + Flows Type = "flows" ) // Valid returns true if the type is a known telemetry type. func (t Type) Valid() bool { switch t { - case Logs, Metrics, Traces: + case Logs, Metrics, Traces, Flows: return true default: return false