From 0020934574b20a25d0254c152f88b155aeee35db Mon Sep 17 00:00:00 2001 From: Louis Date: Wed, 20 Jul 2022 21:27:51 -0700 Subject: [PATCH] Observation domain (#98) * netflow: Add observation domain and point to message The ObservationDomainID and ObservationPointID are two IPFIX fields that identify the entity that is capturing flows and can be used to enrich the context around a specific sample. Parse these fields from the sample and add them to the FlowMessage. Signed-off-by: Adrian Moreno Co-authored-by: Adrian Moreno --- cmd/enricher/pb/flowext.pb.go | 2 +- format/common/text.go | 7 ++- pb/flow.pb.go | 112 ++++++++++++++++++++-------------- pb/flow.proto | 3 + producer/producer_nf.go | 4 ++ producer/producer_nflegacy.go | 4 +- utils/stopper.go | 7 ++- 7 files changed, 89 insertions(+), 50 deletions(-) diff --git a/cmd/enricher/pb/flowext.pb.go b/cmd/enricher/pb/flowext.pb.go index 0e5236b..9aaaa02 100644 --- a/cmd/enricher/pb/flowext.pb.go +++ b/cmd/enricher/pb/flowext.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.26.0 -// protoc v3.17.2 +// protoc v3.21.2 // source: cmd/enricher/pb/flowext.proto package flowpb diff --git a/format/common/text.go b/format/common/text.go index 22bce17..1cbb7ed 100644 --- a/format/common/text.go +++ b/format/common/text.go @@ -3,10 +3,11 @@ package common import ( "encoding/binary" "fmt" - "github.com/golang/protobuf/proto" "net" "reflect" "strings" + + "github.com/golang/protobuf/proto" ) const ( @@ -52,6 +53,8 @@ var ( TextFields = []string{ "Type", + "ObservationPointID", + "ObservationDomainID", "TimeReceived", "SequenceNum", "SamplingRate", @@ -99,6 +102,8 @@ var ( FORMAT_TYPE_INTEGER, FORMAT_TYPE_INTEGER, FORMAT_TYPE_INTEGER, + FORMAT_TYPE_INTEGER, + FORMAT_TYPE_INTEGER, FORMAT_TYPE_IP, FORMAT_TYPE_INTEGER, FORMAT_TYPE_INTEGER, diff --git a/pb/flow.pb.go b/pb/flow.pb.go index 2000a11..613cf97 100644 --- a/pb/flow.pb.go +++ b/pb/flow.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.28.0 -// protoc v3.12.1 +// protoc-gen-go v1.26.0 +// protoc v3.21.2 // source: pb/flow.proto package flowpb @@ -140,16 +140,18 @@ type FlowMessage struct { SrcNet uint32 `protobuf:"varint,16,opt,name=SrcNet,proto3" json:"SrcNet,omitempty"` DstNet uint32 `protobuf:"varint,17,opt,name=DstNet,proto3" json:"DstNet,omitempty"` // MPLS information - HasMPLS bool `protobuf:"varint,53,opt,name=HasMPLS,proto3" json:"HasMPLS,omitempty"` - MPLSCount uint32 `protobuf:"varint,54,opt,name=MPLSCount,proto3" json:"MPLSCount,omitempty"` - MPLS1TTL uint32 `protobuf:"varint,55,opt,name=MPLS1TTL,proto3" json:"MPLS1TTL,omitempty"` // First TTL - MPLS1Label uint32 `protobuf:"varint,56,opt,name=MPLS1Label,proto3" json:"MPLS1Label,omitempty"` // First Label - MPLS2TTL uint32 `protobuf:"varint,57,opt,name=MPLS2TTL,proto3" json:"MPLS2TTL,omitempty"` // Second TTL - MPLS2Label uint32 `protobuf:"varint,58,opt,name=MPLS2Label,proto3" json:"MPLS2Label,omitempty"` // Second Label - MPLS3TTL uint32 `protobuf:"varint,59,opt,name=MPLS3TTL,proto3" json:"MPLS3TTL,omitempty"` // Third TTL - MPLS3Label uint32 `protobuf:"varint,60,opt,name=MPLS3Label,proto3" json:"MPLS3Label,omitempty"` // Third Label - MPLSLastTTL uint32 `protobuf:"varint,61,opt,name=MPLSLastTTL,proto3" json:"MPLSLastTTL,omitempty"` // Last TTL - MPLSLastLabel uint32 `protobuf:"varint,62,opt,name=MPLSLastLabel,proto3" json:"MPLSLastLabel,omitempty"` // Last Label + HasMPLS bool `protobuf:"varint,53,opt,name=HasMPLS,proto3" json:"HasMPLS,omitempty"` + MPLSCount uint32 `protobuf:"varint,54,opt,name=MPLSCount,proto3" json:"MPLSCount,omitempty"` + MPLS1TTL uint32 `protobuf:"varint,55,opt,name=MPLS1TTL,proto3" json:"MPLS1TTL,omitempty"` // First TTL + MPLS1Label uint32 `protobuf:"varint,56,opt,name=MPLS1Label,proto3" json:"MPLS1Label,omitempty"` // First Label + MPLS2TTL uint32 `protobuf:"varint,57,opt,name=MPLS2TTL,proto3" json:"MPLS2TTL,omitempty"` // Second TTL + MPLS2Label uint32 `protobuf:"varint,58,opt,name=MPLS2Label,proto3" json:"MPLS2Label,omitempty"` // Second Label + MPLS3TTL uint32 `protobuf:"varint,59,opt,name=MPLS3TTL,proto3" json:"MPLS3TTL,omitempty"` // Third TTL + MPLS3Label uint32 `protobuf:"varint,60,opt,name=MPLS3Label,proto3" json:"MPLS3Label,omitempty"` // Third Label + MPLSLastTTL uint32 `protobuf:"varint,61,opt,name=MPLSLastTTL,proto3" json:"MPLSLastTTL,omitempty"` // Last TTL + MPLSLastLabel uint32 `protobuf:"varint,62,opt,name=MPLSLastLabel,proto3" json:"MPLSLastLabel,omitempty"` // Last Label + ObservationDomainID uint32 `protobuf:"varint,70,opt,name=ObservationDomainID,proto3" json:"ObservationDomainID,omitempty"` + ObservationPointID uint32 `protobuf:"varint,71,opt,name=ObservationPointID,proto3" json:"ObservationPointID,omitempty"` // Custom allocations CustomInteger1 uint64 `protobuf:"varint,1001,opt,name=CustomInteger1,proto3" json:"CustomInteger1,omitempty"` CustomInteger2 uint64 `protobuf:"varint,1002,opt,name=CustomInteger2,proto3" json:"CustomInteger2,omitempty"` @@ -566,6 +568,20 @@ func (x *FlowMessage) GetMPLSLastLabel() uint32 { return 0 } +func (x *FlowMessage) GetObservationDomainID() uint32 { + if x != nil { + return x.ObservationDomainID + } + return 0 +} + +func (x *FlowMessage) GetObservationPointID() uint32 { + if x != nil { + return x.ObservationPointID + } + return 0 +} + func (x *FlowMessage) GetCustomInteger1() uint64 { if x != nil { return x.CustomInteger1 @@ -640,7 +656,7 @@ var File_pb_flow_proto protoreflect.FileDescriptor var file_pb_flow_proto_rawDesc = []byte{ 0x0a, 0x0d, 0x70, 0x62, 0x2f, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, - 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x22, 0xa6, 0x10, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77, + 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x22, 0x88, 0x11, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x30, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x1c, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x54, @@ -741,40 +757,46 @@ var file_pb_flow_proto_rawDesc = []byte{ 0x28, 0x0d, 0x52, 0x0b, 0x4d, 0x50, 0x4c, 0x53, 0x4c, 0x61, 0x73, 0x74, 0x54, 0x54, 0x4c, 0x12, 0x24, 0x0a, 0x0d, 0x4d, 0x50, 0x4c, 0x53, 0x4c, 0x61, 0x73, 0x74, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x18, 0x3e, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x4d, 0x50, 0x4c, 0x53, 0x4c, 0x61, 0x73, 0x74, - 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, - 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x31, 0x18, 0xe9, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, - 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x31, 0x12, 0x27, - 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x32, - 0x18, 0xea, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, - 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x32, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, - 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x33, 0x18, 0xeb, 0x07, 0x20, 0x01, 0x28, 0x04, - 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x33, + 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x12, 0x30, 0x0a, 0x13, 0x4f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, + 0x74, 0x69, 0x6f, 0x6e, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x49, 0x44, 0x18, 0x46, 0x20, 0x01, + 0x28, 0x0d, 0x52, 0x13, 0x4f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x44, + 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x49, 0x44, 0x12, 0x2e, 0x0a, 0x12, 0x4f, 0x62, 0x73, 0x65, 0x72, + 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x6f, 0x69, 0x6e, 0x74, 0x49, 0x44, 0x18, 0x47, 0x20, + 0x01, 0x28, 0x0d, 0x52, 0x12, 0x4f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, + 0x50, 0x6f, 0x69, 0x6e, 0x74, 0x49, 0x44, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, + 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x31, 0x18, 0xe9, 0x07, 0x20, 0x01, 0x28, 0x04, + 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x31, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, - 0x72, 0x34, 0x18, 0xec, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, - 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x34, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, - 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x35, 0x18, 0xed, 0x07, 0x20, 0x01, + 0x72, 0x32, 0x18, 0xea, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, + 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x32, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, + 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x33, 0x18, 0xeb, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, - 0x72, 0x35, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, - 0x73, 0x31, 0x18, 0xf3, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, - 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x31, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, - 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x18, 0xf4, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, - 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x12, 0x23, 0x0a, 0x0c, - 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x33, 0x18, 0xf5, 0x07, 0x20, - 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, - 0x33, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, - 0x34, 0x18, 0xf6, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, - 0x42, 0x79, 0x74, 0x65, 0x73, 0x34, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, - 0x42, 0x79, 0x74, 0x65, 0x73, 0x35, 0x18, 0xf7, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, - 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x35, 0x22, 0x53, 0x0a, 0x08, 0x46, - 0x6c, 0x6f, 0x77, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x4c, 0x4f, 0x57, 0x55, - 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x46, 0x4c, 0x4f, - 0x57, 0x5f, 0x35, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57, - 0x5f, 0x56, 0x35, 0x10, 0x02, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57, - 0x5f, 0x56, 0x39, 0x10, 0x03, 0x12, 0x09, 0x0a, 0x05, 0x49, 0x50, 0x46, 0x49, 0x58, 0x10, 0x04, - 0x42, 0x29, 0x5a, 0x27, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6e, - 0x65, 0x74, 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x72, 0x2f, 0x67, 0x6f, 0x66, 0x6c, 0x6f, 0x77, - 0x32, 0x2f, 0x70, 0x62, 0x3b, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, - 0x74, 0x6f, 0x33, + 0x72, 0x33, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, + 0x67, 0x65, 0x72, 0x34, 0x18, 0xec, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, + 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x34, 0x12, 0x27, 0x0a, 0x0e, 0x43, + 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x35, 0x18, 0xed, 0x07, + 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, + 0x67, 0x65, 0x72, 0x35, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, + 0x74, 0x65, 0x73, 0x31, 0x18, 0xf3, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, + 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x31, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, + 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x18, 0xf4, 0x07, 0x20, 0x01, 0x28, 0x0c, + 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x12, 0x23, + 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x33, 0x18, 0xf5, + 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, + 0x65, 0x73, 0x33, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, + 0x65, 0x73, 0x34, 0x18, 0xf6, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, + 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x34, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, + 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x35, 0x18, 0xf7, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, + 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x35, 0x22, 0x53, 0x0a, + 0x08, 0x46, 0x6c, 0x6f, 0x77, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x4c, 0x4f, + 0x57, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x46, + 0x4c, 0x4f, 0x57, 0x5f, 0x35, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, + 0x4f, 0x57, 0x5f, 0x56, 0x35, 0x10, 0x02, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, + 0x4f, 0x57, 0x5f, 0x56, 0x39, 0x10, 0x03, 0x12, 0x09, 0x0a, 0x05, 0x49, 0x50, 0x46, 0x49, 0x58, + 0x10, 0x04, 0x42, 0x29, 0x5a, 0x27, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, + 0x2f, 0x6e, 0x65, 0x74, 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x72, 0x2f, 0x67, 0x6f, 0x66, 0x6c, + 0x6f, 0x77, 0x32, 0x2f, 0x70, 0x62, 0x3b, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x62, 0x06, 0x70, + 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/pb/flow.proto b/pb/flow.proto index 0a03483..06c1fa1 100644 --- a/pb/flow.proto +++ b/pb/flow.proto @@ -100,6 +100,9 @@ message FlowMessage { uint32 MPLSLastTTL = 61; // Last TTL uint32 MPLSLastLabel = 62; // Last Label + uint32 ObservationDomainID = 70; + uint32 ObservationPointID = 71; + // Custom fields: start after ID 1000: // uint32 MyCustomField = 1000; diff --git a/producer/producer_nf.go b/producer/producer_nf.go index b556844..0382e83 100644 --- a/producer/producer_nf.go +++ b/producer/producer_nf.go @@ -199,6 +199,9 @@ func ConvertNetFlowDataSet(version uint16, baseTime uint32, uptime uint32, recor switch df.Type { + case netflow.IPFIX_FIELD_observationPointId: + DecodeUNumber(v, &(flowMessage.ObservationPointID)) + // Statistics case netflow.NFV9_FIELD_IN_BYTES: DecodeUNumber(v, &(flowMessage.Bytes)) @@ -560,6 +563,7 @@ func ProcessMessageNetFlowConfig(msgDec interface{}, samplingRateSys SamplingRat for _, fmsg := range flowMessageSet { fmsg.SequenceNum = seqnum fmsg.SamplingRate = uint64(samplingRate) + fmsg.ObservationDomainID = obsDomainId } default: return flowMessageSet, errors.New("Bad NetFlow/IPFIX version") diff --git a/producer/producer_nflegacy.go b/producer/producer_nflegacy.go index 30d7659..84decd1 100644 --- a/producer/producer_nflegacy.go +++ b/producer/producer_nflegacy.go @@ -16,9 +16,9 @@ func ConvertNetFlowLegacyRecord(baseTime uint32, uptime uint32, record netflowle timeDiffFirst := (uptime - record.First) timeDiffLast := (uptime - record.Last) - flowMessage.TimeFlowStart = uint64(baseTime - timeDiffFirst / 1000) + flowMessage.TimeFlowStart = uint64(baseTime - timeDiffFirst/1000) flowMessage.TimeFlowStartMs = uint64(baseTime)*1000 - uint64(timeDiffFirst) - flowMessage.TimeFlowEnd = uint64(baseTime - timeDiffLast / 1000) + flowMessage.TimeFlowEnd = uint64(baseTime - timeDiffLast/1000) flowMessage.TimeFlowEndMs = uint64(baseTime)*1000 - uint64(timeDiffLast) v := make(net.IP, 4) diff --git a/utils/stopper.go b/utils/stopper.go index fa6f10d..153b1bd 100644 --- a/utils/stopper.go +++ b/utils/stopper.go @@ -22,7 +22,12 @@ func (s *stopper) start() error { func (s *stopper) Shutdown() { if s.stopCh != nil { - close(s.stopCh) + select { + case <-s.stopCh: + default: + close(s.stopCh) + } + s.stopCh = nil } }