Skip to content

Commit e73c782

Browse files
mrproliumrproliu
authored
Support analyze multiple protocol in the same connection (#160)
1 parent 17f479e commit e73c782

File tree

9 files changed

+89
-50
lines changed

9 files changed

+89
-50
lines changed

‎pkg/accesslog/collector/protocols/connection.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,8 @@ import (
2929

3030
type PartitionConnection struct {
3131
connectionID, randomID uint64
32-
dataBuffer *buffer.Buffer
33-
protocol map[enums.ConnectionProtocol]bool
32+
dataBuffers map[enums.ConnectionProtocol]*buffer.Buffer
33+
protocol map[enums.ConnectionProtocol]uint64 // protocol with minimal data id
3434
protocolAnalyzer map[enums.ConnectionProtocol]Protocol
3535
protocolMetrics map[enums.ConnectionProtocol]ProtocolMetrics
3636
closed bool
@@ -48,8 +48,8 @@ func (p *PartitionConnection) IsExistProtocol(protocol enums.ConnectionProtocol)
4848
return exist
4949
}
5050

51-
func (p *PartitionConnection) Buffer() *buffer.Buffer {
52-
return p.dataBuffer
51+
func (p *PartitionConnection) Buffer(protocol enums.ConnectionProtocol) *buffer.Buffer {
52+
return p.dataBuffers[protocol]
5353
}
5454

5555
func (p *PartitionConnection) AppendDetail(ctx *common.AccessLogContext, detail events.SocketDetail) {
@@ -58,12 +58,12 @@ func (p *PartitionConnection) AppendDetail(ctx *common.AccessLogContext, detail
5858
forwarder.SendTransferNoProtocolEvent(ctx, detail)
5959
return
6060
}
61-
p.dataBuffer.AppendDetailEvent(detail)
61+
p.dataBuffers[detail.GetProtocol()].AppendDetailEvent(detail)
6262
}
6363

6464
func (p *PartitionConnection) AppendData(data buffer.SocketDataBuffer) {
6565
if p.skipAllDataAnalyze {
6666
return
6767
}
68-
p.dataBuffer.AppendDataEvent(data)
68+
p.dataBuffers[data.Protocol()].AppendDataEvent(data)
6969
}

‎pkg/accesslog/collector/protocols/http1.go

+10-9
Original file line numberDiff line numberDiff line change
@@ -68,18 +68,19 @@ func (p *HTTP1Protocol) GenerateConnection(connectionID, randomID uint64) Protoc
6868

6969
func (p *HTTP1Protocol) Analyze(connection *PartitionConnection, _ *AnalyzeHelper) error {
7070
metrics := connection.Metrics(enums.ConnectionProtocolHTTP).(*HTTP1Metrics)
71+
buf := connection.Buffer(enums.ConnectionProtocolHTTP)
7172
http1Log.Debugf("ready to analyze HTTP/1 protocol data, connection ID: %d, random ID: %d, data len: %d",
72-
metrics.ConnectionID, metrics.RandomID, connection.Buffer().DataLength())
73-
connection.Buffer().ResetForLoopReading()
73+
metrics.ConnectionID, metrics.RandomID, buf.DataLength())
74+
buf.ResetForLoopReading()
7475
for {
75-
if !connection.Buffer().PrepareForReading() {
76+
if !buf.PrepareForReading() {
7677
return nil
7778
}
7879

79-
messageType, err := reader.IdentityMessageType(connection.Buffer())
80+
messageType, err := reader.IdentityMessageType(buf)
8081
if err != nil {
8182
http1Log.Debugf("failed to identity message type, %v", err)
82-
if connection.Buffer().SkipCurrentElement() {
83+
if buf.SkipCurrentElement() {
8384
break
8485
}
8586
continue
@@ -88,19 +89,19 @@ func (p *HTTP1Protocol) Analyze(connection *PartitionConnection, _ *AnalyzeHelpe
8889
var result enums.ParseResult
8990
switch messageType {
9091
case reader.MessageTypeRequest:
91-
result, _ = p.handleRequest(metrics, connection.Buffer())
92+
result, _ = p.handleRequest(metrics, buf)
9293
case reader.MessageTypeResponse:
93-
result, _ = p.handleResponse(metrics, connection.Buffer())
94+
result, _ = p.handleResponse(metrics, buf)
9495
case reader.MessageTypeUnknown:
9596
result = enums.ParseResultSkipPackage
9697
}
9798

9899
finishReading := false
99100
switch result {
100101
case enums.ParseResultSuccess:
101-
finishReading = connection.Buffer().RemoveReadElements()
102+
finishReading = buf.RemoveReadElements()
102103
case enums.ParseResultSkipPackage:
103-
finishReading = connection.Buffer().SkipCurrentElement()
104+
finishReading = buf.SkipCurrentElement()
104105
}
105106

106107
if finishReading {

‎pkg/accesslog/collector/protocols/http2.go

+11-10
Original file line numberDiff line numberDiff line change
@@ -89,19 +89,20 @@ func (r *HTTP2Protocol) GenerateConnection(connectionID, randomID uint64) Protoc
8989

9090
func (r *HTTP2Protocol) Analyze(connection *PartitionConnection, helper *AnalyzeHelper) error {
9191
http2Metrics := connection.Metrics(enums.ConnectionProtocolHTTP2).(*HTTP2Metrics)
92+
buf := connection.Buffer(enums.ConnectionProtocolHTTP2)
9293
http2Log.Debugf("ready to analyze HTTP/2 protocol data, connection ID: %d, random ID: %d",
9394
http2Metrics.connectionID, http2Metrics.randomID)
94-
connection.Buffer().ResetForLoopReading()
95+
buf.ResetForLoopReading()
9596
for {
96-
if !connection.Buffer().PrepareForReading() {
97+
if !buf.PrepareForReading() {
9798
return nil
9899
}
99100

100-
startPosition := connection.Buffer().Position()
101-
header, err := http2.ReadFrameHeader(connection.Buffer())
101+
startPosition := buf.Position()
102+
header, err := http2.ReadFrameHeader(buf)
102103
if err != nil {
103104
http2Log.Debugf("failed to read frame header, %v", err)
104-
if connection.Buffer().SkipCurrentElement() {
105+
if buf.SkipCurrentElement() {
105106
break
106107
}
107108
continue
@@ -112,12 +113,12 @@ func (r *HTTP2Protocol) Analyze(connection *PartitionConnection, helper *Analyze
112113
var result enums.ParseResult
113114
switch header.Type {
114115
case http2.FrameHeaders:
115-
result, protocolBreak, _ = r.handleHeader(&header, startPosition, http2Metrics, connection.Buffer())
116+
result, protocolBreak, _ = r.handleHeader(&header, startPosition, http2Metrics, buf)
116117
case http2.FrameData:
117-
result, protocolBreak, _ = r.handleData(&header, startPosition, http2Metrics, connection.Buffer())
118+
result, protocolBreak, _ = r.handleData(&header, startPosition, http2Metrics, buf)
118119
default:
119120
tmp := make([]byte, header.Length)
120-
if err := connection.Buffer().ReadUntilBufferFull(tmp); err != nil {
121+
if err := buf.ReadUntilBufferFull(tmp); err != nil {
121122
if errors.Is(err, buffer.ErrNotComplete) {
122123
result = enums.ParseResultSkipPackage
123124
} else {
@@ -139,9 +140,9 @@ func (r *HTTP2Protocol) Analyze(connection *PartitionConnection, helper *Analyze
139140
finishReading := false
140141
switch result {
141142
case enums.ParseResultSuccess:
142-
finishReading = connection.Buffer().RemoveReadElements()
143+
finishReading = buf.RemoveReadElements()
143144
case enums.ParseResultSkipPackage:
144-
finishReading = connection.Buffer().SkipCurrentElement()
145+
finishReading = buf.SkipCurrentElement()
145146
}
146147

147148
if finishReading {

‎pkg/accesslog/collector/protocols/queue.go

+43-20
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"errors"
2323
"fmt"
2424
"os"
25+
"sort"
2526
"sync"
2627
"time"
2728

@@ -119,25 +120,30 @@ type PartitionContext struct {
119120
analyzeLocker sync.Mutex
120121
}
121122

122-
func newPartitionConnection(protocolMgr *ProtocolManager, conID, randomID uint64, protocol enums.ConnectionProtocol) *PartitionConnection {
123+
func newPartitionConnection(protocolMgr *ProtocolManager, conID, randomID uint64,
124+
protocol enums.ConnectionProtocol, currentDataID uint64) *PartitionConnection {
123125
connection := &PartitionConnection{
124126
connectionID: conID,
125127
randomID: randomID,
126-
dataBuffer: buffer.NewBuffer(),
127-
protocol: make(map[enums.ConnectionProtocol]bool),
128+
dataBuffers: make(map[enums.ConnectionProtocol]*buffer.Buffer),
129+
protocol: make(map[enums.ConnectionProtocol]uint64),
128130
protocolAnalyzer: make(map[enums.ConnectionProtocol]Protocol),
129131
protocolMetrics: make(map[enums.ConnectionProtocol]ProtocolMetrics),
130132
}
131-
connection.appendProtocolIfNeed(protocolMgr, conID, randomID, protocol)
133+
connection.appendProtocolIfNeed(protocolMgr, conID, randomID, protocol, currentDataID)
132134
return connection
133135
}
134136

135-
func (p *PartitionConnection) appendProtocolIfNeed(protocolMgr *ProtocolManager, conID, randomID uint64, protocol enums.ConnectionProtocol) {
136-
if _, exist := p.protocol[protocol]; !exist {
137+
func (p *PartitionConnection) appendProtocolIfNeed(protocolMgr *ProtocolManager, conID, randomID uint64,
138+
protocol enums.ConnectionProtocol, currentDataID uint64) {
139+
if minDataID, exist := p.protocol[protocol]; !exist {
137140
analyzer := protocolMgr.GetProtocol(protocol)
138-
p.protocol[protocol] = true
141+
p.protocol[protocol] = currentDataID
142+
p.dataBuffers[protocol] = buffer.NewBuffer()
139143
p.protocolAnalyzer[protocol] = analyzer
140144
p.protocolMetrics[protocol] = analyzer.GenerateConnection(conID, randomID)
145+
} else if currentDataID < minDataID {
146+
p.protocol[protocol] = currentDataID
141147
}
142148
}
143149

@@ -212,26 +218,27 @@ func (p *PartitionContext) Consume(data interface{}) {
212218
forwarder.SendTransferNoProtocolEvent(p.context, event)
213219
return
214220
}
215-
connection := p.getConnectionContext(event.GetConnectionID(), event.GetRandomID(), event.GetProtocol())
221+
connection := p.getConnectionContext(event.GetConnectionID(), event.GetRandomID(), event.GetProtocol(), event.DataID())
216222
connection.AppendDetail(p.context, event)
217223
case *events.SocketDataUploadEvent:
218224
pid, _ := events.ParseConnectionID(event.ConnectionID)
219225
log.Debugf("receive the socket data event, connection ID: %d, random ID: %d, pid: %d, data id: %d, sequence: %d, protocol: %d",
220-
event.ConnectionID, event.RandomID, pid, event.DataID0, event.Sequence0, event.Protocol)
221-
connection := p.getConnectionContext(event.ConnectionID, event.RandomID, event.Protocol)
226+
event.ConnectionID, event.RandomID, pid, event.DataID0, event.Sequence0, event.Protocol0)
227+
connection := p.getConnectionContext(event.ConnectionID, event.RandomID, event.Protocol0, event.DataID0)
222228
connection.AppendData(event)
223229
}
224230
}
225231

226-
func (p *PartitionContext) getConnectionContext(connectionID, randomID uint64, protocol enums.ConnectionProtocol) *PartitionConnection {
232+
func (p *PartitionContext) getConnectionContext(connectionID, randomID uint64,
233+
protocol enums.ConnectionProtocol, currentDataID uint64) *PartitionConnection {
227234
conKey := p.buildConnectionKey(connectionID, randomID)
228235
conn, exist := p.connections.Get(conKey)
229236
if exist {
230237
connection := conn.(*PartitionConnection)
231-
connection.appendProtocolIfNeed(p.protocolMgr, connectionID, randomID, protocol)
238+
connection.appendProtocolIfNeed(p.protocolMgr, connectionID, randomID, protocol, currentDataID)
232239
return connection
233240
}
234-
result := newPartitionConnection(p.protocolMgr, connectionID, randomID, protocol)
241+
result := newPartitionConnection(p.protocolMgr, connectionID, randomID, protocol, currentDataID)
235242
p.connections.Set(conKey, result)
236243
return result
237244
}
@@ -254,7 +261,10 @@ func (p *PartitionContext) processEvents() {
254261
p.processConnectionEvents(info)
255262

256263
// if the connection already closed and not contains any buffer data, then delete the connection
257-
bufLen := info.dataBuffer.DataLength()
264+
var bufLen = 0
265+
for _, buf := range info.dataBuffers {
266+
bufLen += buf.DataLength()
267+
}
258268
if bufLen > 0 {
259269
return
260270
}
@@ -309,9 +319,11 @@ func (p *PartitionContext) processExpireEvents() {
309319
}
310320

311321
func (p *PartitionContext) processConnectionExpireEvents(connection *PartitionConnection) {
312-
if c := connection.dataBuffer.DeleteExpireEvents(maxBufferExpireDuration); c > 0 {
313-
log.Debugf("total removed %d expired socket data events from connection ID: %d, random ID: %d", c,
314-
connection.connectionID, connection.randomID)
322+
for _, buf := range connection.dataBuffers {
323+
if c := buf.DeleteExpireEvents(maxBufferExpireDuration); c > 0 {
324+
log.Debugf("total removed %d expired socket data events from connection ID: %d, random ID: %d", c,
325+
connection.connectionID, connection.randomID)
326+
}
315327
}
316328
}
317329

@@ -320,8 +332,17 @@ func (p *PartitionContext) processConnectionEvents(connection *PartitionConnecti
320332
return
321333
}
322334
helper := &AnalyzeHelper{}
323-
for protocol, analyzer := range connection.protocolAnalyzer {
324-
if err := analyzer.Analyze(connection, helper); err != nil {
335+
336+
// since the socket data/detail are getting unsorted, so rover need to using the minimal data id to analyze to ensure the order
337+
sortedProtocols := make([]enums.ConnectionProtocol, 0, len(connection.protocol))
338+
for protocol := range connection.protocol {
339+
sortedProtocols = append(sortedProtocols, protocol)
340+
}
341+
sort.Slice(sortedProtocols, func(i, j int) bool {
342+
return connection.protocol[sortedProtocols[i]] < connection.protocol[sortedProtocols[j]]
343+
})
344+
for _, protocol := range sortedProtocols {
345+
if err := connection.protocolAnalyzer[protocol].Analyze(connection, helper); err != nil {
325346
log.Warnf("failed to analyze the %s protocol data: %v", enums.ConnectionProtocolString(protocol), err)
326347
}
327348
}
@@ -330,6 +351,8 @@ func (p *PartitionContext) processConnectionEvents(connection *PartitionConnecti
330351
// notify the connection manager to skip analyze all data(just sending the detail)
331352
connection.skipAllDataAnalyze = true
332353
p.context.ConnectionMgr.SkipAllDataAnalyze(connection.connectionID, connection.randomID)
333-
connection.dataBuffer.Clean()
354+
for _, buf := range connection.dataBuffers {
355+
buf.Clean()
356+
}
334357
}
335358
}

‎pkg/accesslog/events/data.go

+5-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ import (
2424
)
2525

2626
type SocketDataUploadEvent struct {
27-
Protocol enums.ConnectionProtocol
27+
Protocol0 enums.ConnectionProtocol
2828
HaveReduce uint8
2929
Direction0 enums.SocketDataDirection
3030
Finished uint8
@@ -39,6 +39,10 @@ type SocketDataUploadEvent struct {
3939
Buffer [2048]byte
4040
}
4141

42+
func (s *SocketDataUploadEvent) Protocol() enums.ConnectionProtocol {
43+
return s.Protocol0
44+
}
45+
4246
func (s *SocketDataUploadEvent) GenerateConnectionID() string {
4347
return fmt.Sprintf("%d_%d", s.ConnectionID, s.RandomID)
4448
}

‎pkg/profiling/task/network/analyze/events/data.go

+5-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ import (
2424
)
2525

2626
type SocketDataUploadEvent struct {
27-
Protocol enums.ConnectionProtocol
27+
Protocol0 enums.ConnectionProtocol
2828
HaveReduce uint8
2929
Direction0 enums.SocketDataDirection
3030
Finished uint8
@@ -39,6 +39,10 @@ type SocketDataUploadEvent struct {
3939
Buffer [2048]byte
4040
}
4141

42+
func (s *SocketDataUploadEvent) Protocol() enums.ConnectionProtocol {
43+
return s.Protocol0
44+
}
45+
4246
func (s *SocketDataUploadEvent) GenerateConnectionID() string {
4347
return fmt.Sprintf("%d_%d", s.ConnectionID, s.RandomID)
4448
}

‎pkg/profiling/task/network/analyze/layer7/protocols/protocols.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -71,10 +71,10 @@ func (a *Analyzer) Start(ctx context.Context) {
7171
}
7272

7373
func (a *Analyzer) ReceiveSocketDataEvent(event *events.SocketDataUploadEvent) {
74-
analyzer := a.protocols[event.Protocol]
74+
analyzer := a.protocols[event.Protocol()]
7575
if analyzer == nil {
7676
log.Warnf("could not found any protocol to handle socket data, connection id: %s, protocol: %s(%d)",
77-
event.GenerateConnectionID(), enums.ConnectionProtocolString(event.Protocol), event.Protocol)
77+
event.GenerateConnectionID(), enums.ConnectionProtocolString(event.Protocol()), event.Protocol())
7878
return
7979
}
8080
analyzer.ReceiveSocketData(a.ctx, event)

‎pkg/tools/buffer/buffer.go

+6
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,8 @@ var (
3333
)
3434

3535
type SocketDataBuffer interface {
36+
// Protocol of the buffer
37+
Protocol() enums.ConnectionProtocol
3638
// GenerateConnectionID for identity the buffer belong which connection
3739
GenerateConnectionID() string
3840
// BufferData of the buffer
@@ -88,6 +90,10 @@ type SocketDataEventLimited struct {
8890
Size int
8991
}
9092

93+
func (s *SocketDataEventLimited) Protocol() enums.ConnectionProtocol {
94+
return s.SocketDataBuffer.Protocol()
95+
}
96+
9197
func (s *SocketDataEventLimited) BufferData() []byte {
9298
return s.SocketDataBuffer.BufferData()[s.From:s.Size]
9399
}

‎pkg/tools/ssl/gotls.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -206,7 +206,7 @@ func (r *Register) generateGOTLSSymbolOffsets(register *Register, elfFile *elf.F
206206

207207
sym := register.SearchSymbol(func(a, b string) bool {
208208
return a == b
209-
}, "go.itab.*net.TCPConn,net.Conn")
209+
}, "go.itab.*net.TCPConn,net.Conn", "go:itab.*net.TCPConn,net.Conn")
210210
if sym == nil {
211211
log.Warnf("could not found the tcp connection symbol: go.itab.*net.TCPConn,net.Conn")
212212
return nil, nil

0 commit comments

Comments
 (0)