paulmataruso/goflow2
1
0
Fork 0
mirror of https://github.com/openobserve/goflow2.git synced 2025-10-31 12:03:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

mapping: allow lists to be mapped (#149)

Browse Source
This commit is contained in:
Louis
2023-03-28 21:55:21 -07:00
committed by GitHub
parent 5529d4993b
commit 40e5ef08fd
6 changed files with 83 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/goflow2/mapping.yaml
View File

@@ -6,6 +6,10 @@ ipfix:
destination: CustomInteger2
# penprovided: false
# pen: 0
- field: 137
destination: CustomList_1
penprovided: true
pen: 2636
netflowv9:
mapping:
- field: 7

11
format/common/text.go
View File

@@ -220,6 +220,17 @@ func FormatMessageReflectCustom(msg interface{}, ext, quotes, sep, sign string,
switch fieldValue.Kind() {
case reflect.String:
fstr[i] = fmt.Sprintf("%s%s%s%s%q", quotes, s, quotes, sign, fieldValue.Interface())
case reflect.Slice:
c := fieldValue.Len()
v := "["
for i := 0; i < c; i++ {
v += fmt.Sprintf("%v", fieldValue.Index(i).Interface())
if i < c-1 {
v += ","
}
}
v += "]"
fstr[i] = fmt.Sprintf("%s%s%s%s%s", quotes, s, quotes, sign, v)
default:
fstr[i] = fmt.Sprintf("%s%s%s%s%v", quotes, s, quotes, sign, fieldValue.Interface())
}

50
pb/flow.pb.go
View File

@@ -158,16 +158,17 @@ type FlowMessage struct {
ObservationDomainId uint32 `protobuf:"varint,70,opt,name=observation_domain_id,json=observationDomainId,proto3" json:"observation_domain_id,omitempty"`
ObservationPointId uint32 `protobuf:"varint,71,opt,name=observation_point_id,json=observationPointId,proto3" json:"observation_point_id,omitempty"`
// Custom allocations
CustomInteger_1 uint64 `protobuf:"varint,1001,opt,name=custom_integer_1,json=customInteger1,proto3" json:"custom_integer_1,omitempty"`
CustomInteger_2 uint64 `protobuf:"varint,1002,opt,name=custom_integer_2,json=customInteger2,proto3" json:"custom_integer_2,omitempty"`
CustomInteger_3 uint64 `protobuf:"varint,1003,opt,name=custom_integer_3,json=customInteger3,proto3" json:"custom_integer_3,omitempty"`
CustomInteger_4 uint64 `protobuf:"varint,1004,opt,name=custom_integer_4,json=customInteger4,proto3" json:"custom_integer_4,omitempty"`
CustomInteger_5 uint64 `protobuf:"varint,1005,opt,name=custom_integer_5,json=customInteger5,proto3" json:"custom_integer_5,omitempty"`
CustomBytes_1 []byte `protobuf:"bytes,1011,opt,name=custom_bytes_1,json=customBytes1,proto3" json:"custom_bytes_1,omitempty"`
CustomBytes_2 []byte `protobuf:"bytes,1012,opt,name=custom_bytes_2,json=customBytes2,proto3" json:"custom_bytes_2,omitempty"`
CustomBytes_3 []byte `protobuf:"bytes,1013,opt,name=custom_bytes_3,json=customBytes3,proto3" json:"custom_bytes_3,omitempty"`
CustomBytes_4 []byte `protobuf:"bytes,1014,opt,name=custom_bytes_4,json=customBytes4,proto3" json:"custom_bytes_4,omitempty"`
CustomBytes_5 []byte `protobuf:"bytes,1015,opt,name=custom_bytes_5,json=customBytes5,proto3" json:"custom_bytes_5,omitempty"`
CustomInteger_1 uint64 `protobuf:"varint,1001,opt,name=custom_integer_1,json=customInteger1,proto3" json:"custom_integer_1,omitempty"`
CustomInteger_2 uint64 `protobuf:"varint,1002,opt,name=custom_integer_2,json=customInteger2,proto3" json:"custom_integer_2,omitempty"`
CustomInteger_3 uint64 `protobuf:"varint,1003,opt,name=custom_integer_3,json=customInteger3,proto3" json:"custom_integer_3,omitempty"`
CustomInteger_4 uint64 `protobuf:"varint,1004,opt,name=custom_integer_4,json=customInteger4,proto3" json:"custom_integer_4,omitempty"`
CustomInteger_5 uint64 `protobuf:"varint,1005,opt,name=custom_integer_5,json=customInteger5,proto3" json:"custom_integer_5,omitempty"`
CustomBytes_1 []byte `protobuf:"bytes,1011,opt,name=custom_bytes_1,json=customBytes1,proto3" json:"custom_bytes_1,omitempty"`
CustomBytes_2 []byte `protobuf:"bytes,1012,opt,name=custom_bytes_2,json=customBytes2,proto3" json:"custom_bytes_2,omitempty"`
CustomBytes_3 []byte `protobuf:"bytes,1013,opt,name=custom_bytes_3,json=customBytes3,proto3" json:"custom_bytes_3,omitempty"`
CustomBytes_4 []byte `protobuf:"bytes,1014,opt,name=custom_bytes_4,json=customBytes4,proto3" json:"custom_bytes_4,omitempty"`
CustomBytes_5 []byte `protobuf:"bytes,1015,opt,name=custom_bytes_5,json=customBytes5,proto3" json:"custom_bytes_5,omitempty"`
CustomList_1 []uint32 `protobuf:"varint,1021,rep,packed,name=custom_list_1,json=customList1,proto3" json:"custom_list_1,omitempty"`
}
func (x *FlowMessage) Reset() {
@@ -685,11 +686,18 @@ func (x *FlowMessage) GetCustomBytes_5() []byte {
return nil
}
func (x *FlowMessage) GetCustomList_1() []uint32 {
if x != nil {
return x.CustomList_1
}
return nil
}
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, 0xeb, 0x12, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77,
0x06, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x22, 0x90, 0x13, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77,
0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x30, 0x0a, 0x04, 0x74, 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,
@@ -835,15 +843,17 @@ var file_pb_flow_proto_rawDesc = []byte{
0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x34, 0x12, 0x25, 0x0a, 0x0e, 0x63, 0x75, 0x73,
0x74, 0x6f, 0x6d, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x5f, 0x35, 0x18, 0xf7, 0x07, 0x20, 0x01,
0x28, 0x0c, 0x52, 0x0c, 0x63, 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,
0x12, 0x23, 0x0a, 0x0d, 0x63, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x5f, 0x6c, 0x69, 0x73, 0x74, 0x5f,
0x31, 0x18, 0xfd, 0x07, 0x20, 0x03, 0x28, 0x0d, 0x52, 0x0b, 0x63, 0x75, 0x73, 0x74, 0x6f, 0x6d,
0x4c, 0x69, 0x73, 0x74, 0x31, 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 (

2
pb/flow.proto
View File

@@ -126,4 +126,6 @@ message FlowMessage {
bytes custom_bytes_4 = 1014;
bytes custom_bytes_5 = 1015;
repeated uint32 custom_list_1 = 1021;
}

32
producer/reflect.go
View File

@@ -41,6 +41,14 @@ func GetBytes(d []byte, offset int, length int) []byte {
return chunk
}
func IsUInt(k reflect.Kind) bool {
return k == reflect.Uint8 || k == reflect.Uint16 || k == reflect.Uint32 || k == reflect.Uint64
}
func IsInt(k reflect.Kind) bool {
return k == reflect.Int8 || k == reflect.Int16 || k == reflect.Int32 || k == reflect.Int64
}
func MapCustomNetFlow(flowMessage *flowmessage.FlowMessage, df netflow.DataField, mapper *NetFlowMapper) {
if mapper == nil {
return
@@ -62,11 +70,27 @@ func MapCustom(flowMessage *flowmessage.FlowMessage, v []byte, destination strin
typeDest := fieldValue.Type()
fieldValueAddr := fieldValue.Addr()
if typeDest.Kind() == reflect.Slice && typeDest.Elem().Kind() == reflect.Uint8 {
fieldValue.SetBytes(v)
} else if fieldValueAddr.IsValid() && (typeDest.Kind() == reflect.Uint8 || typeDest.Kind() == reflect.Uint16 || typeDest.Kind() == reflect.Uint32 || typeDest.Kind() == reflect.Uint64) {
if typeDest.Kind() == reflect.Slice {
if typeDest.Elem().Kind() == reflect.Uint8 {
fieldValue.SetBytes(v)
} else {
item := reflect.New(typeDest.Elem())
if IsUInt(typeDest.Elem().Kind()) {
DecodeUNumber(v, item.Interface())
} else if IsUInt(typeDest.Elem().Kind()) {
DecodeUNumber(v, item.Interface())
}
itemi := reflect.Indirect(item)
tmpFieldValue := reflect.Append(fieldValue, itemi)
fieldValue.Set(tmpFieldValue)
}
} else if fieldValueAddr.IsValid() && IsUInt(typeDest.Kind()) {
DecodeUNumber(v, fieldValueAddr.Interface())
} else if fieldValueAddr.IsValid() && (typeDest.Kind() == reflect.Int8 || typeDest.Kind() == reflect.Int16 || typeDest.Kind() == reflect.Int32 || typeDest.Kind() == reflect.Int64) {
} else if fieldValueAddr.IsValid() && IsInt(typeDest.Kind()) {
DecodeNumber(v, fieldValueAddr.Interface())
}
}

16
utils/utils.go
View File

@@ -71,20 +71,20 @@ type Formatter interface {
type DefaultLogTransport struct {
}
func (s *DefaultLogTransport) Publish(msgs []*flowmessage.FlowMessage) {
for _, msg := range msgs {
fmt.Printf("%v\n", FlowMessageToString(msg))
func (s *DefaultLogTransport) Publish(msgs []*flowmessage.FlowMessage) {
for _, msg := range msgs {
fmt.Printf("%v\n", FlowMessageToString(msg))
}
}
}
type DefaultJSONTransport struct {
}
func (s *DefaultJSONTransport) Publish(msgs []*flowmessage.FlowMessage) {
for _, msg := range msgs {
fmt.Printf("%v\n", FlowMessageToJSON(msg))
func (s *DefaultJSONTransport) Publish(msgs []*flowmessage.FlowMessage) {
for _, msg := range msgs {
fmt.Printf("%v\n", FlowMessageToJSON(msg))
}
}
}
*/
type DefaultErrorCallback struct {
Logger Logger