Skip to content

Commit cdacb22

Browse files
rachelScoutrachelScout
committed
add third parameter domainID/srcID to key in template cache for IPFix/v9
1 parent 40e66a7 commit cdacb22

File tree

10 files changed

+170
-530
lines changed

10 files changed

+170
-530
lines changed

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ The IPFIX data decodes to JSON format and IDs are [IANA IPFIX element ID](http:/
3434

3535
## Decoded sFlow data
3636
```json
37-
{"Version":5,"IPVersion":1,"AgentSubID":5,"SequenceNo":37591,"SysUpTime":3287084017,"SamplesNo":1,"Samples":[{"SequenceNo":1530345639,"SourceID":0,"SamplingRate":4096,"SamplePool":1938456576,"Drops":0,"Input":536,"Output":728,"RecordsNo":3,"Records":{"ExtRouter":{"NextHop":"115.131.251.90","SrcMask":24,"DstMask":14},"ExtSwitch":{"SrcVlan":0,"SrcPriority":0,"DstVlan":0,"DstPriority":0},"RawHeader":{"L2":{"SrcMAC":"58:00:bb:e7:57:6f","DstMAC":"f4:a7:39:44:a8:27","Vlan":0,"EtherType":2048},"L3":{"Version":4,"TOS":0,"TotalLen":1452,"ID":13515,"Flags":0,"FragOff":0,"TTL":62,"Protocol":6,"Checksum":8564,"Src":"10.1.8.5","Dst":"161.140.24.181"},"L4":{"SrcPort":443,"DstPort":56521,"DataOffset":5,"Reserved":0,"Flags":16}}}}],"IPAddress":"192.168.10.0","ColTime": 1646157296}
37+
{"Version":5,"IPVersion":1,"AgentSubID":5,"SequenceNo":37591,"SysUpTime":3287084017,"SamplesNo":1,"Samples":[{"SequenceNo":1530345639,"SourceID":0,"SamplingRate":4096,"SamplePool":1938456576,"Drops":0,"Input":536,"Output":728,"RecordsNo":3,"Records":{"ExtRouter":{"NextHop":"115.131.251.90","SrcMask":24,"DstMask":14},"ExtSwitch":{"SrcVlan":0,"SrcPriority":0,"DstVlan":0,"DstPriority":0},"RawHeader":{"L2":{"SrcMAC":"58:00:bb:e7:57:6f","DstMAC":"f4:a7:39:44:a8:27","Vlan":0,"EtherType":2048},"L3":{"Version":4,"TOS":0,"TotalLen":1452,"ID":13515,"Flags":0,"FragOff":0,"TTL":62,"Protocol":6,"Checksum":8564,"Src":"10.1.8.5","Dst":"161.140.24.181"},"L4":{"SrcPort":443,"DstPort":56521,"DataOffset":5,"Reserved":0,"Flags":16}}}}],"IPAddress":"192.168.10.0"}
3838
```
3939
## Decoded Netflow v5 data
4040
``` json

‎go.mod‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -3,16 +3,16 @@ module github.com/EdgeCast/vflow
33
go 1.15
44

55
require (
6-
github.com/ClickHouse/clickhouse-go v1.5.4
7-
github.com/Shopify/sarama v1.33.0
6+
github.com/ClickHouse/clickhouse-go v1.4.3
7+
github.com/Shopify/sarama v1.26.3
88
github.com/bsm/sarama-cluster v2.1.15+incompatible
9-
github.com/nats-io/nats-server/v2 v2.8.2 // indirect
10-
github.com/nats-io/nats.go v1.15.0
11-
github.com/nsqio/go-nsq v1.1.0
12-
github.com/onsi/ginkgo v1.16.5 // indirect
13-
github.com/onsi/gomega v1.19.0 // indirect
14-
github.com/prometheus/client_golang v1.12.2
15-
github.com/segmentio/kafka-go v0.4.31
16-
golang.org/x/net v0.0.0-20220513224357-95641704303c
17-
gopkg.in/yaml.v2 v2.4.0
9+
github.com/nats-io/nats-server/v2 v2.1.8 // indirect
10+
github.com/nats-io/nats.go v1.10.0
11+
github.com/nsqio/go-nsq v1.0.8
12+
github.com/onsi/ginkgo v1.14.2 // indirect
13+
github.com/onsi/gomega v1.10.3 // indirect
14+
github.com/prometheus/client_golang v1.6.0
15+
github.com/segmentio/kafka-go v0.4.7
16+
golang.org/x/net v0.0.0-20201021035429-f5854403a974
17+
gopkg.in/yaml.v2 v2.3.0
1818
)

‎go.sum‎

Lines changed: 78 additions & 481 deletions
Large diffs are not rendered by default.

‎ipfix/decoder.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -164,17 +164,19 @@ func (d *Decoder) decodeSet(mem MemCache, msg *Message) error {
164164
// This check is somewhat redundant with the switch-clause below, but the retrieve() operation should not be executed inside the loop.
165165
if setHeader.SetID > 255 {
166166
var ok bool
167-
if tr, ok = mem.retrieve(setHeader.SetID, d.raddr); !ok {
167+
if tr, ok = mem.retrieve(setHeader.SetID, d.raddr, msg.Header.DomainID); !ok {
168168
select {
169169
case rpcChan <- RPCRequest{
170170
ID: setHeader.SetID,
171171
IP: d.raddr,
172+
SrcID: msg.Header.DomainID,
172173
}:
173174
default:
174175
}
175-
err = nonfatalError{fmt.Errorf("%s unknown ipfix template id# %d",
176+
err = nonfatalError{fmt.Errorf("%s unknown ipfix template id# %d with domain ID %d",
176177
d.raddr.String(),
177178
setHeader.SetID,
179+
msg.Header.DomainID,
178180
)}
179181
}
180182
}
@@ -196,7 +198,7 @@ func (d *Decoder) decodeSet(mem MemCache, msg *Message) error {
196198
err = tr.unmarshalOpts(d.reader)
197199
}
198200
if err == nil {
199-
mem.insert(tr.TemplateID, d.raddr, tr)
201+
mem.insert(tr.TemplateID, d.raddr, tr, msg.Header.DomainID)
200202
}
201203
} else if setID >= 4 && setID <= 255 {
202204
// Reserved set, do not read any records

‎ipfix/decoder_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ import (
3131
var tpl, optsTpl, multiMessage, unknownDatasetMessage []byte
3232

3333
func init() {
34-
// IPFIX packet including template SetID:2, 25 fields
34+
// IPFIX packet including template SetID:2, 25 fields, Domain id 33792
3535
tpl = []byte{
3636
0x0, 0xa, 0x0, 0x7c, 0x58, 0x90, 0xd6, 0x40, 0x28, 0xf7,
3737
0xa0, 0x4a, 0x0, 0x0, 0x84, 0x0, 0x0, 0x2, 0x0, 0x6c, 0x1,
@@ -205,8 +205,8 @@ func TestUnknownDatasetsMessage(t *testing.T) {
205205
t.Error("Did not expect any result datasets, but got", l)
206206
}
207207
expectedErrorStr := `Multiple errors:
208-
- 127.0.0.1 unknown ipfix template id# 264
209-
- 127.0.0.1 unknown ipfix template id# 264`
208+
- 127.0.0.1 unknown ipfix template id# 264 with domain ID 1
209+
- 127.0.0.1 unknown ipfix template id# 264 with domain ID 1`
210210
if err == nil || err.Error() != expectedErrorStr {
211211
t.Error("Received unexpected erorr:", err)
212212
}

‎ipfix/memcache.go‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -79,27 +79,32 @@ func GetCache(cacheFile string) MemCache {
7979
return m
8080
}
8181

82-
func (m MemCache) getShard(id uint16, addr net.IP) (*TemplatesShard, uint32) {
83-
b := make([]byte, 2)
84-
binary.BigEndian.PutUint16(b, id)
85-
key := append(addr, b...)
86-
82+
func (m MemCache) getShard(templateId uint16, addr net.IP, domainId uint32) (*TemplatesShard, uint32) {
83+
var key []byte
8784
hash := fnv.New32()
85+
dId := make([]byte, 4)
86+
tID := make([]byte, 2)
87+
binary.LittleEndian.PutUint32(dId, domainId)
88+
binary.BigEndian.PutUint16(tID, templateId)
89+
key = append(key, addr...)
90+
key = append(key, dId...)
91+
key = append(key, tID...)
92+
8893
hash.Write(key)
8994
hSum32 := hash.Sum32()
9095

9196
return m[uint(hSum32)%uint(shardNo)], hSum32
9297
}
9398

94-
func (m MemCache) insert(id uint16, addr net.IP, tr TemplateRecord) {
95-
shard, key := m.getShard(id, addr)
99+
func (m MemCache) insert(id uint16, addr net.IP, tr TemplateRecord, domainID uint32) {
100+
shard, key := m.getShard(id, addr, domainID)
96101
shard.Lock()
97102
defer shard.Unlock()
98103
shard.Templates[key] = Data{tr, time.Now().Unix()}
99104
}
100105

101-
func (m MemCache) retrieve(id uint16, addr net.IP) (TemplateRecord, bool) {
102-
shard, key := m.getShard(id, addr)
106+
func (m MemCache) retrieve(id uint16, addr net.IP, domainID uint32) (TemplateRecord, bool) {
107+
shard, key := m.getShard(id, addr, domainID)
103108
shard.RLock()
104109
defer shard.RUnlock()
105110
v, ok := shard.Templates[key]

‎ipfix/memcache_rpc.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ type RPCConfig struct {
5858
type RPCRequest struct {
5959
ID uint16
6060
IP net.IP
61+
SrcID uint32
6162
}
6263

6364
type vFlowServer struct {
@@ -91,7 +92,7 @@ func NewRPC(mCache MemCache) *IRPC {
9192
func (r *IRPC) Get(req RPCRequest, resp *TemplateRecord) error {
9293
var ok bool
9394

94-
*resp, ok = r.mCache.retrieve(req.ID, req.IP)
95+
*resp, ok = r.mCache.retrieve(req.ID, req.IP, req.SrcID)
9596
if !ok {
9697
return errNotAvail
9798
}
@@ -168,7 +169,7 @@ func RPC(m MemCache, config *RPCConfig) {
168169
continue
169170
}
170171

171-
m.insert(req.ID, req.IP, *tr)
172+
m.insert(req.ID, req.IP, *tr, req.SrcID)
172173
break
173174
}
174175

‎ipfix/memcache_test.go‎

Lines changed: 36 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ func TestMemCacheRetrieve(t *testing.T) {
3333
mCache := GetCache("cache.file")
3434
d := NewDecoder(ip, tpl)
3535
d.Decode(mCache)
36-
v, ok := mCache.retrieve(256, ip)
36+
v, ok := mCache.retrieve(256, ip, 33792)
3737
if !ok {
3838
t.Error("expected mCache retrieve status true, got", ok)
3939
}
@@ -48,9 +48,9 @@ func TestMemCacheInsert(t *testing.T) {
4848
mCache := GetCache("cache.file")
4949

5050
tpl.TemplateID = 310
51-
mCache.insert(310, ip, tpl)
51+
mCache.insert(310, ip, tpl, 513)
5252

53-
v, ok := mCache.retrieve(310, ip)
53+
v, ok := mCache.retrieve(310, ip, 513)
5454
if !ok {
5555
t.Error("expected mCache retrieve status true, got", ok)
5656
}
@@ -65,15 +65,45 @@ func TestMemCacheAllSetIds(t *testing.T) {
6565
mCache := GetCache("cache.file")
6666

6767
tpl.TemplateID = 310
68-
mCache.insert(tpl.TemplateID, ip, tpl)
68+
mCache.insert(tpl.TemplateID, ip, tpl, 513)
6969
tpl.TemplateID = 410
70-
mCache.insert(tpl.TemplateID, ip, tpl)
70+
mCache.insert(tpl.TemplateID, ip, tpl, 513)
7171
tpl.TemplateID = 210
72-
mCache.insert(tpl.TemplateID, ip, tpl)
72+
mCache.insert(tpl.TemplateID, ip, tpl, 513)
7373

7474
expected := []int{210, 310, 410}
7575
actual := mCache.allSetIds()
7676
if !reflect.DeepEqual(expected, actual) {
7777
t.Errorf("Expected set IDs %v, got %v", expected, actual)
7878
}
7979
}
80+
81+
func TestMemCache_keyWithDifferentDomainIDs(t *testing.T) {
82+
var tpl TemplateRecord
83+
ip := net.ParseIP("127.0.0.1")
84+
mCache := GetCache("cache.file")
85+
86+
tpl.TemplateID = 310
87+
tpl.FieldCount = 19
88+
mCache.insert(tpl.TemplateID, ip, tpl, 513)
89+
tpl.FieldCount = 21
90+
mCache.insert(tpl.TemplateID, ip, tpl, 514)
91+
92+
v, ok := mCache.retrieve(tpl.TemplateID, ip, 513)
93+
94+
if !ok {
95+
t.Error("expected mCache retrieve status true, got", ok)
96+
}
97+
if v.FieldCount != 19 {
98+
t.Error("expected template id#:310 with Field count#:19, got", v.TemplateID, v.FieldCount)
99+
}
100+
101+
v, ok = mCache.retrieve(tpl.TemplateID, ip, 514)
102+
103+
if !ok {
104+
t.Error("expected mCache retrieve status true, got", ok)
105+
}
106+
if v.FieldCount != 21 {
107+
t.Error("expected template id#:310 with Field count#:21, got", v.TemplateID, v.FieldCount)
108+
}
109+
}

‎netflow/v9/decoder.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -426,11 +426,11 @@ func (d *Decoder) decodeSet(mem MemCache, msg *Message) error {
426426
// This check is somewhat redundant with the switch-clause below, but the retrieve() operation should not be executed inside the loop.
427427
if setHeader.FlowSetID > 255 {
428428
var ok bool
429-
tr, ok = mem.retrieve(setHeader.FlowSetID, d.raddr)
429+
tr, ok = mem.retrieve(setHeader.FlowSetID, d.raddr, msg.Header.SrcID)
430430
if !ok {
431-
err = nonfatalError(fmt.Errorf("%s unknown netflow template id# %d",
431+
err = nonfatalError(fmt.Errorf("%s unknown netflow template id# %d from sourceID %d",
432432
d.raddr.String(),
433-
setHeader.FlowSetID,
433+
setHeader.FlowSetID, msg.Header.SrcID,
434434
))
435435
}
436436
}
@@ -446,9 +446,9 @@ func (d *Decoder) decodeSet(mem MemCache, msg *Message) error {
446446
err = tr.unmarshalOpts(d.reader)
447447
}
448448
if err == nil {
449-
mem.insert(tr.TemplateID, d.raddr, tr)
449+
mem.insert(tr.TemplateID, d.raddr, tr, msg.Header.SrcID)
450450
}
451-
} else if setId >= 4 && setId <= 255 {
451+
} else if setId <= 255 {
452452
// Reserved set, do not read any records
453453
break
454454
} else {

‎netflow/v9/memcache.go‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -78,27 +78,32 @@ func GetCache(cacheFile string) MemCache {
7878
return m
7979
}
8080

81-
func (m MemCache) getShard(id uint16, addr net.IP) (*TemplatesShard, uint32) {
82-
b := make([]byte, 2)
83-
binary.BigEndian.PutUint16(b, id)
84-
key := append(addr, b...)
85-
81+
func (m MemCache) getShard(templateId uint16, addr net.IP, srcId uint32) (*TemplatesShard, uint32) {
82+
var key []byte
8683
hash := fnv.New32()
84+
sId := make([]byte, 4)
85+
tID := make([]byte, 2)
86+
binary.LittleEndian.PutUint32(sId, srcId)
87+
binary.BigEndian.PutUint16(tID, templateId)
88+
key = append(key, addr...)
89+
key = append(key, sId...)
90+
key = append(key, tID...)
91+
8792
hash.Write(key)
8893
hSum32 := hash.Sum32()
8994

9095
return m[uint(hSum32)%uint(shardNo)], hSum32
9196
}
9297

93-
func (m *MemCache) insert(id uint16, addr net.IP, tr TemplateRecord) {
94-
shard, key := m.getShard(id, addr)
98+
func (m *MemCache) insert(id uint16, addr net.IP, tr TemplateRecord, srcID uint32) {
99+
shard, key := m.getShard(id, addr, srcID)
95100
shard.Lock()
96101
defer shard.Unlock()
97102
shard.Templates[key] = Data{tr, time.Now().Unix()}
98103
}
99104

100-
func (m *MemCache) retrieve(id uint16, addr net.IP) (TemplateRecord, bool) {
101-
shard, key := m.getShard(id, addr)
105+
func (m *MemCache) retrieve(id uint16, addr net.IP, srcID uint32) (TemplateRecord, bool) {
106+
shard, key := m.getShard(id, addr, srcID)
102107
shard.RLock()
103108
defer shard.RUnlock()
104109
v, ok := shard.Templates[key]

0 commit comments

Comments
 (0)