Skip to content

Commit

Permalink
remove element map from record (antrea-io#206)
Browse files Browse the repository at this point in the history
  • Loading branch information
@zyiou
zyiou authored Jun 4, 2021
1 parent 67fed69 commit 0ca84f4
Show file tree
Hide file tree
Showing 7 changed files with 88 additions and 109 deletions.
75 changes: 41 additions & 34 deletions pkg/entities/record.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import (

type Record interface {
PrepareRecord() error
AddInfoElement(element *InfoElementWithValue, isDecoding bool) error
AddInfoElement(element *InfoElementWithValue) error
// TODO: Functions for multiple elements as well.
GetBuffer() []byte
GetTemplateID() uint16
Expand All @@ -41,50 +41,54 @@ type Record interface {

type baseRecord struct {
buffer []byte
len int
fieldCount uint16
templateID uint16
orderedElementList []*InfoElementWithValue
elementsMap map[string]*InfoElementWithValue
isDecoding bool
Record
}

type dataRecord struct {
baseRecord
}

func NewDataRecord(id uint16, numElements int) *dataRecord {
return &dataRecord{
func NewDataRecord(id uint16, numElements int, isDecoding bool) *dataRecord {
record := &dataRecord{
baseRecord{
buffer: make([]byte, 0),
len: 0,
fieldCount: 0,
templateID: id,
orderedElementList: make([]*InfoElementWithValue, 0),
elementsMap: make(map[string]*InfoElementWithValue, numElements),
buffer: make([]byte, 0),
fieldCount: 0,
templateID: id,
isDecoding: isDecoding,
},
}
if isDecoding {
record.orderedElementList = make([]*InfoElementWithValue, numElements)
}
return record
}

type templateRecord struct {
baseRecord
// Minimum data record length required to be sent for this template.
// Elements with variable length are considered to be one byte.
minDataRecLength uint16
// index is used when adding elements to orderedElementList
index int
}

func NewTemplateRecord(id uint16, numElements int) *templateRecord {
return &templateRecord{
func NewTemplateRecord(id uint16, numElements int, isDecoding bool) *templateRecord {
record := &templateRecord{
baseRecord{
buffer: make([]byte, 0),
len: 0,
fieldCount: uint16(numElements),
templateID: id,
orderedElementList: make([]*InfoElementWithValue, 0),
elementsMap: make(map[string]*InfoElementWithValue, numElements),
buffer: make([]byte, 0),
fieldCount: uint16(numElements),
templateID: id,
isDecoding: isDecoding,
},
0,
0,
}
record.orderedElementList = make([]*InfoElementWithValue, numElements)
return record
}

func (b *baseRecord) GetBuffer() []byte {
Expand All @@ -99,43 +103,46 @@ func (b *baseRecord) GetFieldCount() uint16 {
return b.fieldCount
}

func (d *baseRecord) GetOrderedElementList() []*InfoElementWithValue {
return d.orderedElementList
func (b *baseRecord) GetOrderedElementList() []*InfoElementWithValue {
return b.orderedElementList
}

func (b *baseRecord) GetInfoElementWithValue(name string) (*InfoElementWithValue, bool) {
if element, exist := b.elementsMap[name]; exist {
return element, exist
} else {
return nil, false
for _, element := range b.orderedElementList {
if element.Element.Name == name {
return element, true
}
}
return nil, false
}

func (d *dataRecord) PrepareRecord() error {
// We do not have to do anything if it is data record
return nil
}

func (d *dataRecord) AddInfoElement(element *InfoElementWithValue, isDecoding bool) error {
d.fieldCount++
func (d *dataRecord) AddInfoElement(element *InfoElementWithValue) error {
var value interface{}
var err error
if isDecoding {
if d.isDecoding {
value, err = DecodeToIEDataType(element.Element.DataType, element.Value)
if err != nil {
return err
}
element.Value = value
if len(d.orderedElementList) <= int(d.fieldCount) {
d.orderedElementList = append(d.orderedElementList, element)
} else {
d.orderedElementList[d.fieldCount] = element
}
} else {
buffBytes, err := EncodeToIEDataType(element.Element.DataType, element.Value)
if err != nil {
return err
}
d.buffer = append(d.buffer, buffBytes...)
}

d.orderedElementList = append(d.orderedElementList, element)
d.elementsMap[element.Element.Name] = element
d.fieldCount++
return nil
}

Expand All @@ -150,7 +157,7 @@ func (t *templateRecord) PrepareRecord() error {
return nil
}

func (t *templateRecord) AddInfoElement(element *InfoElementWithValue, isDecoding bool) error {
func (t *templateRecord) AddInfoElement(element *InfoElementWithValue) error {
// val could be used to specify smaller length than default? For now assert it to be nil
if element.Value != nil {
return fmt.Errorf("AddInfoElement(templateRecord) cannot take value %v (nil is expected)", element.Value)
Expand All @@ -170,8 +177,8 @@ func (t *templateRecord) AddInfoElement(element *InfoElementWithValue, isDecodin
binary.BigEndian.PutUint32(addBytes, element.Element.EnterpriseId)
t.buffer = append(t.buffer, addBytes...)
}
t.orderedElementList = append(t.orderedElementList, element)
t.elementsMap[element.Element.Name] = element
t.orderedElementList[t.index] = element
t.index++
// Keep track of minimum data record length required for sanity check
if element.Element.Len == VariableLength {
t.minDataRecLength = t.minDataRecLength + 1
Expand Down
24 changes: 12 additions & 12 deletions pkg/entities/record_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,8 @@ func TestPrepareRecord(t *testing.T) {
expectLen uint16
expectErr error
}{
{NewDataRecord(uniqueTemplateID, 1), 0, nil},
{NewTemplateRecord(uniqueTemplateID, 1), 4, nil},
{NewDataRecord(uniqueTemplateID, 1, false), 0, nil},
{NewTemplateRecord(uniqueTemplateID, 1, false), 4, nil},
}

for _, test := range prepareRecordTests {
Expand Down Expand Up @@ -80,8 +80,8 @@ func TestAddInfoElements(t *testing.T) {
ieList []*InfoElement
valList []interface{}
}{
{NewTemplateRecord(uniqueTemplateID, 1), testIEs, nil},
{NewDataRecord(uniqueTemplateID, len(testIEs)), testIEs, valData},
{NewTemplateRecord(uniqueTemplateID, 12, false), testIEs, nil},
{NewDataRecord(uniqueTemplateID, len(testIEs), false), testIEs, valData},
}

for i, test := range addIETests {
Expand All @@ -90,11 +90,11 @@ func TestAddInfoElements(t *testing.T) {
if i == 0 {
// For template record
ie := NewInfoElementWithValue(testIE, nil)
actualErr = test.record.AddInfoElement(ie, false)
actualErr = test.record.AddInfoElement(ie)
} else {
// For data record
ie := NewInfoElementWithValue(testIE, test.valList[j])
actualErr = test.record.AddInfoElement(ie, false)
actualErr = test.record.AddInfoElement(ie)
if testIE.Len == VariableLength {
_, ok := test.valList[j].(string)
if !ok {
Expand All @@ -114,18 +114,18 @@ func TestAddInfoElements(t *testing.T) {
}

func TestGetInfoElementWithValue(t *testing.T) {
templateRec := NewTemplateRecord(256, 1)
templateRec.elementsMap = make(map[string]*InfoElementWithValue)
templateRec := NewTemplateRecord(256, 1, true)
templateRec.orderedElementList = make([]*InfoElementWithValue, 0)
ie := NewInfoElementWithValue(NewInfoElement("sourceIPv4Address", 8, 18, 0, 4), nil)
templateRec.elementsMap["sourceIPv4Address"] = ie
templateRec.orderedElementList = append(templateRec.orderedElementList, ie)
_, exist := templateRec.GetInfoElementWithValue("sourceIPv4Address")
assert.Equal(t, true, exist)
_, exist = templateRec.GetInfoElementWithValue("destinationIPv4Address")
assert.Equal(t, false, exist)
dataRec := NewDataRecord(256, 1)
dataRec.elementsMap = make(map[string]*InfoElementWithValue)
dataRec := NewDataRecord(256, 1, true)
dataRec.orderedElementList = make([]*InfoElementWithValue, 0)
ie = NewInfoElementWithValue(NewInfoElement("sourceIPv4Address", 8, 18, 0, 4), net.ParseIP("10.0.0.1"))
dataRec.elementsMap["sourceIPv4Address"] = ie
dataRec.orderedElementList = append(dataRec.orderedElementList, ie)
infoElementWithValue, _ := dataRec.GetInfoElementWithValue("sourceIPv4Address")
assert.Equal(t, net.ParseIP("10.0.0.1"), infoElementWithValue.Value)
infoElementWithValue, _ = dataRec.GetInfoElementWithValue("destinationIPv4Address")
Expand Down
6 changes: 3 additions & 3 deletions pkg/entities/set.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,9 +106,9 @@ func (s *set) UpdateLenInHeader() {
func (s *set) AddRecord(elements []*InfoElementWithValue, templateID uint16) error {
var record Record
if s.setType == Data {
record = NewDataRecord(templateID, len(elements))
record = NewDataRecord(templateID, len(elements), s.isDecoding)
} else if s.setType == Template {
record = NewTemplateRecord(templateID, len(elements))
record = NewTemplateRecord(templateID, len(elements), s.isDecoding)
err := record.PrepareRecord()
if err != nil {
return err
Expand All @@ -118,7 +118,7 @@ func (s *set) AddRecord(elements []*InfoElementWithValue, templateID uint16) err
}

for _, element := range elements {
err := record.AddInfoElement(element, s.isDecoding)
err := record.AddInfoElement(element)
if err != nil {
return err
}
Expand Down
17 changes: 5 additions & 12 deletions pkg/entities/set_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,7 @@ func TestAddRecordIPv4Addresses(t *testing.T) {
elements = append(elements, ie1, ie2)
err = encodingSet.AddRecord(elements, 256)
assert.NoError(t, err)
infoElementWithValue, _ := encodingSet.GetRecords()[0].GetInfoElementWithValue("sourceIPv4Address")
assert.Equal(t, net.IP([]byte{0xa, 0x0, 0x0, 0x1}), infoElementWithValue.Value)
infoElementWithValue, _ = encodingSet.GetRecords()[0].GetInfoElementWithValue("destinationIPv4Address")
assert.Equal(t, net.IP([]byte{0xa, 0x0, 0x0, 0x2}), infoElementWithValue.Value)
assert.Equal(t, []byte{0xa, 0x0, 0x0, 0x1, 0xa, 0x0, 0x0, 0x2}, encodingSet.GetRecords()[0].GetBuffer())
}

func TestAddRecordIPv6Addresses(t *testing.T) {
Expand All @@ -52,10 +49,7 @@ func TestAddRecordIPv6Addresses(t *testing.T) {
err := newSet.PrepareSet(Template, testTemplateID)
assert.NoError(t, err)
newSet.AddRecord(elements, 256)
_, exist := newSet.GetRecords()[0].GetInfoElementWithValue("sourceIPv6Address")
assert.Equal(t, true, exist)
_, exist = newSet.GetRecords()[0].GetInfoElementWithValue("destinationIPv6Address")
assert.Equal(t, true, exist)
assert.Equal(t, []byte{0x1, 0x0, 0x0, 0x2, 0x0, 0x1b, 0x0, 0x10, 0x0, 0x1c, 0x0, 0x10}, newSet.GetRecords()[0].GetBuffer())
newSet.ResetSet()
// Test with data record
err = newSet.PrepareSet(Data, testTemplateID)
Expand All @@ -65,10 +59,9 @@ func TestAddRecordIPv6Addresses(t *testing.T) {
ie2 = NewInfoElementWithValue(NewInfoElement("destinationIPv6Address", 28, 19, 0, 16), net.ParseIP("2001:0:3238:DFE1:63::FEFC"))
elements = append(elements, ie1, ie2)
newSet.AddRecord(elements, 256)
infoElementWithValue, _ := newSet.GetRecords()[0].GetInfoElementWithValue("sourceIPv6Address")
assert.Equal(t, net.IP([]byte{0x20, 0x1, 0x0, 0x0, 0x32, 0x38, 0xdf, 0xe1, 0x0, 0x63, 0x0, 0x0, 0x0, 0x0, 0xfe, 0xfb}), infoElementWithValue.Value)
infoElementWithValue, _ = newSet.GetRecords()[0].GetInfoElementWithValue("destinationIPv6Address")
assert.Equal(t, net.IP([]byte{0x20, 0x1, 0x0, 0x0, 0x32, 0x38, 0xdf, 0xe1, 0x0, 0x63, 0x0, 0x0, 0x0, 0x0, 0xfe, 0xfc}), infoElementWithValue.Value)
srcIP := []byte(net.ParseIP("2001:0:3238:DFE1:63::FEFB"))
dstIP := []byte(net.ParseIP("2001:0:3238:DFE1:63::FEFC"))
assert.Equal(t, append(srcIP, dstIP...), newSet.GetRecords()[0].GetBuffer())
}

func TestGetSetType(t *testing.T) {
Expand Down
8 changes: 4 additions & 4 deletions pkg/entities/testing/mock_record.go
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

42 changes: 20 additions & 22 deletions pkg/intermediate/aggregate.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -130,13 +130,13 @@ func (a *AggregationProcess) Stop() {

// AggregateMsgByFlowKey gets flow key from records in message and stores in cache
func (a *AggregationProcess) AggregateMsgByFlowKey(message *entities.Message) error {
if err := addOriginalExporterInfo(message); err != nil {
return err
}
set := message.GetSet()
if set.GetSetType() == entities.Template { // skip template records
if set.GetSetType() != entities.Data { // only process data records
return nil
}
if err := addOriginalExporterInfo(message); err != nil {
return err
}
records := set.GetRecords()
invalidRecs := 0
for _, record := range records {
Expand Down Expand Up @@ -568,7 +568,7 @@ func (a *AggregationProcess) addFieldsForStatsAggregation(record entities.Record
buffBytes := make([]byte, 8)
binary.BigEndian.PutUint64(buffBytes, uint64(0))
ieWithValue := entities.NewInfoElementWithValue(ie, buffBytes)
err = record.AddInfoElement(ieWithValue, true)
err = record.AddInfoElement(ieWithValue)
if err != nil {
return err
}
Expand Down Expand Up @@ -733,18 +733,18 @@ func addOriginalExporterInfo(message *entities.Message) error {
return err
}

if set.GetSetType() == entities.Template {
originalExporterIP = entities.NewInfoElementWithValue(ie, nil)
} else if set.GetSetType() == entities.Data {
if isIPv4 {
originalExporterIP = entities.NewInfoElementWithValue(ie, net.ParseIP(message.GetExportAddress()).To4())
} else {
originalExporterIP = entities.NewInfoElementWithValue(ie, net.ParseIP(message.GetExportAddress()).To16())
}
var value []byte
if isIPv4 {
value, err = entities.EncodeToIEDataType(entities.Ipv4Address, net.ParseIP(message.GetExportAddress()).To4())
} else {
return fmt.Errorf("set type %d is not supported", set.GetSetType())
value, err = entities.EncodeToIEDataType(entities.Ipv6Address, net.ParseIP(message.GetExportAddress()).To16())
}
err = record.AddInfoElement(originalExporterIP, false)
if err != nil {
return fmt.Errorf("error when encoding originalExporterIP: %v", err)
}
originalExporterIP = entities.NewInfoElementWithValue(ie, value)

err = record.AddInfoElement(originalExporterIP)
if err != nil {
return err
}
Expand All @@ -754,14 +754,12 @@ func addOriginalExporterInfo(message *entities.Message) error {
if err != nil {
return fmt.Errorf("IANA Registry is not loaded correctly with originalObservationDomainId")
}
if set.GetSetType() == entities.Template {
originalObservationDomainId = entities.NewInfoElementWithValue(ie, nil)
} else if set.GetSetType() == entities.Data {
originalObservationDomainId = entities.NewInfoElementWithValue(ie, message.GetObsDomainID())
} else {
return fmt.Errorf("set type %d is not supported", set.GetSetType())
value, err = entities.EncodeToIEDataType(entities.Unsigned32, message.GetObsDomainID())
if err != nil {
return fmt.Errorf("error when encoding originalObservationDomainId: %v", err)
}
err = record.AddInfoElement(originalObservationDomainId, false)
originalObservationDomainId = entities.NewInfoElementWithValue(ie, value)
err = record.AddInfoElement(originalObservationDomainId)
if err != nil {
return err
}
Expand Down
Loading

0 comments on commit 0ca84f4

Please sign in to comment.