From bbdfb9e07e43748906c386617e242d78c009a765 Mon Sep 17 00:00:00 2001 From: Alexander Lopintsev Date: Tue, 2 Dec 2025 17:02:37 +0300 Subject: [PATCH 1/2] Add support for packetInterval and packetSpace to configure the sampling rate. --- producer/proto/producer_nf.go | 51 ++------------------ producer/proto/sampling_rate.go | 82 +++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 47 deletions(-) create mode 100644 producer/proto/sampling_rate.go diff --git a/producer/proto/producer_nf.go b/producer/proto/producer_nf.go index 4f95f777..fbd0c227 100644 --- a/producer/proto/producer_nf.go +++ b/producer/proto/producer_nf.go @@ -1,14 +1,11 @@ package protoproducer import ( - "bytes" "encoding/binary" "fmt" "sync" - "time" "github.com/netsampler/goflow2/v2/decoders/netflow" - "github.com/netsampler/goflow2/v2/decoders/utils" flowmessage "github.com/netsampler/goflow2/v2/pb" "github.com/netsampler/goflow2/v2/producer" ) @@ -69,41 +66,6 @@ func (s *SingleSamplingRateSystem) GetSamplingRate(version uint16, obsDomainId u return s.Sampling, nil } -func NetFlowLookFor(dataFields []netflow.DataField, typeId uint16) (bool, interface{}) { - for _, dataField := range dataFields { - if dataField.Type == typeId { - return true, dataField.Value - } - } - return false, nil -} - -func NetFlowPopulate(dataFields []netflow.DataField, typeId uint16, addr interface{}) (bool, error) { - exists, value := NetFlowLookFor(dataFields, typeId) - if exists && value != nil { - valueBytes, ok := value.([]byte) - valueReader := bytes.NewBuffer(valueBytes) - if ok { - switch addrt := addr.(type) { - //case *(net.IP): - // *addrt = valueBytes - case *(time.Time): - t := uint64(0) - if err := utils.BinaryRead(valueReader, binary.BigEndian, &t); err != nil { - return false, err - } - t64 := int64(t / 1000) - *addrt = time.Unix(t64, 0) - default: - if err := utils.BinaryRead(valueReader, binary.BigEndian, addr); err != nil { - return false, err - } - } - } - } - return exists, nil -} - func WriteUDecoded(o uint64, out interface{}) error { switch t := out.(type) { case *byte: @@ -671,18 +633,13 @@ func SearchNetFlowDataSets(version uint16, baseTime uint32, uptime uint32, dataF func SearchNetFlowOptionDataSets(dataFlowSet []netflow.OptionsDataFlowSet) (samplingRate uint32, found bool, err error) { for _, dataFlowSetItem := range dataFlowSet { for _, record := range dataFlowSetItem.Records { - if found, err := NetFlowPopulate(record.OptionsValues, 305, &samplingRate); err != nil || found { - return samplingRate, found, err - } - if found, err := NetFlowPopulate(record.OptionsValues, 50, &samplingRate); err != nil || found { - return samplingRate, found, err - } - if found, err := NetFlowPopulate(record.OptionsValues, 34, &samplingRate); err != nil || found { - return samplingRate, found, err + samplingOptions, err := SamplingFromOptions(record) + if err == nil { + return samplingOptions.SamplingRate(), true, nil } } } - return samplingRate, found, err + return 0, false, nil } func SplitNetFlowSets(packetNFv9 netflow.NFv9Packet) ([]netflow.DataFlowSet, []netflow.TemplateFlowSet, []netflow.NFv9OptionsTemplateFlowSet, []netflow.OptionsDataFlowSet) { diff --git a/producer/proto/sampling_rate.go b/producer/proto/sampling_rate.go new file mode 100644 index 00000000..0f9d5d3c --- /dev/null +++ b/producer/proto/sampling_rate.go @@ -0,0 +1,82 @@ +package protoproducer + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" + + "github.com/netsampler/goflow2/v2/decoders/netflow" + "github.com/netsampler/goflow2/v2/decoders/utils" +) + +type SamplingRateOptions struct { + PacketSpace uint32 + PacketInterval uint32 + RandomInterval uint32 + SamplingInterval uint32 +} + +func SamplingFromOptions(records netflow.OptionsDataRecord) (*SamplingRateOptions, error) { + samplingOptions := SamplingRateOptions{} + for _, option := range records.OptionsValues { + switch option.Type { + case netflow.IPFIX_FIELD_samplingPacketSpace: + if err := binDecode(option.Value, &samplingOptions.PacketSpace); err != nil { + return nil, fmt.Errorf("can't make samplingPacketSpace from option value %b", option.Value) + } + case netflow.IPFIX_FIELD_samplingPacketInterval: + if err := binDecode(option.Value, &samplingOptions.PacketInterval); err != nil { + return nil, fmt.Errorf("can't make samplingPacketInterval from option value %b", option.Value) + } + case netflow.IPFIX_FIELD_samplerRandomInterval: + if err := binDecode(option.Value, &samplingOptions.RandomInterval); err != nil { + return nil, fmt.Errorf("can't make samplerRandomInterval from option value %b", option.Value) + } + case netflow.IPFIX_FIELD_samplingInterval: + if err := binDecode(option.Value, &samplingOptions.SamplingInterval); err != nil { + return nil, fmt.Errorf("can't make samplingInterval from option value %b", option.Value) + } + default: + } + } + if samplingOptions.SamplingRate() > 0 { + return &samplingOptions, nil + } + return nil, errors.New("no sampling rate options found") +} + +func (s *SamplingRateOptions) SamplingRate() uint32 { + var tmpRate uint32 + if s.PacketInterval > 0 { + tmpRate += s.PacketInterval + if s.PacketSpace > 0 { + tmpRate += s.PacketSpace + } + return tmpRate + } + if s.RandomInterval > 0 { + return s.RandomInterval + } + if s.SamplingInterval > 0 { + return s.SamplingInterval + } + return 0 +} + +func binDecode(source any, destination any) error { + if source == nil { + return errors.New("can't decode nil source") + } + if destination == nil { + return errors.New("can't decode to nil destination") + } + valueBytes, ok := source.([]byte) + if !ok { + return errors.New("source value is not []byte") + } + if err := utils.BinaryRead(bytes.NewBuffer(valueBytes), binary.BigEndian, destination); err != nil { + return err + } + return nil +} From f1fdcf55ee1d76f879a5b0f57ae1c22676727b31 Mon Sep 17 00:00:00 2001 From: Alexander Lopintsev Date: Mon, 15 Dec 2025 16:42:07 +0300 Subject: [PATCH 2/2] Restore public functions NetFlowPopulate and NetFlowLookFor --- producer/proto/producer_nf.go | 38 +++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/producer/proto/producer_nf.go b/producer/proto/producer_nf.go index fbd0c227..f46a0706 100644 --- a/producer/proto/producer_nf.go +++ b/producer/proto/producer_nf.go @@ -1,11 +1,14 @@ package protoproducer import ( + "bytes" "encoding/binary" "fmt" "sync" + "time" "github.com/netsampler/goflow2/v2/decoders/netflow" + "github.com/netsampler/goflow2/v2/decoders/utils" flowmessage "github.com/netsampler/goflow2/v2/pb" "github.com/netsampler/goflow2/v2/producer" ) @@ -66,6 +69,41 @@ func (s *SingleSamplingRateSystem) GetSamplingRate(version uint16, obsDomainId u return s.Sampling, nil } +func NetFlowLookFor(dataFields []netflow.DataField, typeId uint16) (bool, interface{}) { + for _, dataField := range dataFields { + if dataField.Type == typeId { + return true, dataField.Value + } + } + return false, nil +} + +func NetFlowPopulate(dataFields []netflow.DataField, typeId uint16, addr interface{}) (bool, error) { + exists, value := NetFlowLookFor(dataFields, typeId) + if exists && value != nil { + valueBytes, ok := value.([]byte) + valueReader := bytes.NewBuffer(valueBytes) + if ok { + switch addrt := addr.(type) { + //case *(net.IP): + // *addrt = valueBytes + case *(time.Time): + t := uint64(0) + if err := utils.BinaryRead(valueReader, binary.BigEndian, &t); err != nil { + return false, err + } + t64 := int64(t / 1000) + *addrt = time.Unix(t64, 0) + default: + if err := utils.BinaryRead(valueReader, binary.BigEndian, addr); err != nil { + return false, err + } + } + } + } + return exists, nil +} + func WriteUDecoded(o uint64, out interface{}) error { switch t := out.(type) { case *byte: