feat: support rocketMQ #231
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
|
@hengyoush I am currently encountering problems while executing, and there may be issues with the changes I made. Can you help me take a look? sudo ./kyanos watch rocketmq --debug-output rocketmq --remote-ports 9876,8080 --protocol-log-level 5
ERRO[0000] load gotls uprobe failed: load program: invalid argument:
0: R1=ctx() R10=fp0
; int probe_entry_tls_conn_read(struct pt_regs* ctx) {
0: (bf) r8 = r1 ; R1=ctx() R8_w=ctx()
; uint64_t id = bpf_get_current_pid_tgid();
1: (85) call bpf_get_current_pid_tgid#14 ; R0_w=scalar()
; uint32_t tgid = id >> 32;
2: (77) r0 >>= 32 ; R0_w=scalar(smin=0,smax=umax=0xffffffff,var_off=(0x0; 0xffffffff))
; uint32_t tgid = id >> 32;
|
|
I forgot to regenerate the BPF directory. |
|
Have you modified the eBPF code? 
this bug has been fixed in main branch, you should update your branch. |
yes，I need try to regenerate it. |
…oush#221) * docs: introduce prettier and md-padding to format docs Signed-off-by: spencercjh <spencercjh@gmail.com> * fix: make github alerts work Signed-off-by: spencercjh <spencercjh@gmail.com> * fix: make all markdown extensions work Signed-off-by: spencercjh <spencercjh@gmail.com> * fix: reformat new codes from main Signed-off-by: spencercjh <spencercjh@gmail.com> --------- Signed-off-by: spencercjh <spencercjh@gmail.com>
…d in server side (hengyoush#232) fix: add fallback logic to calculate totaltime when nicin event missed in server side (hengyoush#232)
|
@hengyoush After updated and still have a problem, did my changes affect it? sudo ./kyanos watch |
|
No worries. I think your changes should not cause this error, this error is only the gotls probe loading failure, will not affect the function you are developing now, I will check this problem later. @XmchxUp What your kernel version? |
|
|
I tested your branch on my ubuntu24.04(kernel version: static __inline void assign_arg(void* arg, size_t arg_size, struct location_t loc, const void* sp,
uint64_t* regs) {
if (loc.type == kLocationTypeStack) {
bpf_probe_read(arg, arg_size, sp + loc.offset);
} else if (loc.type == kLocationTypeRegisters) {
if (loc.offset >= 0) {
if (loc.offset <= (1<<10)) {
bpf_probe_read(arg, arg_size, (char*)regs + loc.offset);
}
}
}
} |
agent/protocol/rocketmq/rocketmq.go
Outdated
| } | ||
|
|
||
| func (r *RocketMQMessage) FormatToString() string { | ||
| return fmt.Sprintf("base=[%s] command=[%s] payload=[%s]", r.FrameBase.String(), "todo", r.Body) |
There was a problem hiding this comment.
Is the TODO here expected?
I mean because it's a draft PR
agent/protocol/rocketmq/rocketmq.go
Outdated
| if !ok { | ||
| return protocol.ParseResult{ | ||
| ParseState: protocol.Ignore, | ||
| ReadBytes: frameSize, | ||
| } | ||
| } else { | ||
| message.FrameBase = fb | ||
| return protocol.ParseResult{ | ||
| ParseState: protocol.Success, | ||
| ReadBytes: frameSize, | ||
| ParsedMessages: []protocol.ParsedMessage{message}, | ||
| } | ||
| } |
There was a problem hiding this comment.
It can be simplified
| if !ok { | |
| return protocol.ParseResult{ | |
| ParseState: protocol.Ignore, | |
| ReadBytes: frameSize, | |
| } | |
| } else { | |
| message.FrameBase = fb | |
| return protocol.ParseResult{ | |
| ParseState: protocol.Success, | |
| ReadBytes: frameSize, | |
| ParsedMessages: []protocol.ParsedMessage{message}, | |
| } | |
| } | |
| if !ok { | |
| return protocol.ParseResult{ | |
| ParseState: protocol.Ignore, | |
| ReadBytes: frameSize, | |
| } | |
| } | |
| message.FrameBase = fb | |
| return protocol.ParseResult{ | |
| ParseState: protocol.Success, | |
| ReadBytes: frameSize, | |
| ParsedMessages: []protocol.ParsedMessage{message}, | |
| } |
agent/protocol/rocketmq/rocketmq.go
Outdated
| } | ||
|
|
||
| func (parser *RocketMQStreamParser) parseHeader(headerBody []byte, serializedType byte) (*RocketMQMessage, error) { | ||
| fmt.Println(serializedType) |
There was a problem hiding this comment.
This one looks alike a left behind debug
agent/protocol/rocketmq/rocketmq.go
Outdated
| func (parser *RocketMQStreamParser) parseHeader(headerBody []byte, serializedType byte) (*RocketMQMessage, error) { | ||
| fmt.Println(serializedType) | ||
| message := &RocketMQMessage{} | ||
| if serializedType == 0 { |
There was a problem hiding this comment.
Return the error currently in the else here if serializedType != 0
And simplify the dozen line if statement
agent/protocol/rocketmq/rocketmq.go
Outdated
| if len(buffer) < 8 { | ||
| return protocol.ParseResult{ | ||
| ParseState: protocol.NeedsMoreData, | ||
| } | ||
| } | ||
|
|
||
| frameSize := int(binary.BigEndian.Uint32(buffer[:4])) | ||
| if frameSize > len(buffer) { | ||
| return protocol.ParseResult{ParseState: protocol.NeedsMoreData} | ||
| } | ||
|
|
||
| headerLength := binary.BigEndian.Uint32(buffer[4:8]) | ||
| headerDataLen := headerLength & 0xFFFFFF | ||
| serializedType := byte((headerLength >> 24) & 0xFF) | ||
|
|
||
| if len(buffer) < 8+int(headerDataLen) { |
There was a problem hiding this comment.
Thr 4 and 8 are magical numbers here. I would like to suggest you to add constants
agent/protocol/rocketmq/rocketmq.go
Outdated
| if len(headerBody) < 18 { | ||
| return nil, errors.New("invalid header size") | ||
| } | ||
|
|
||
| message.RequestCode = int16(binary.BigEndian.Uint16(headerBody[:2])) | ||
| message.LanguageFlag = headerBody[2] | ||
| message.VersionFlag = int16(binary.BigEndian.Uint16(headerBody[3:5])) | ||
| message.Opaque = int32(binary.BigEndian.Uint32(headerBody[5:9])) | ||
| message.RequestFlag = int32(binary.BigEndian.Uint32(headerBody[9:13])) | ||
| message.RemarkLength = int32(binary.BigEndian.Uint32(headerBody[13:17])) | ||
|
|
||
| if int(message.RemarkLength) > len(headerBody[17:]) { | ||
| return nil, errors.New("invalid remark length") | ||
| } | ||
| message.Remark = headerBody[17 : 17+message.RemarkLength] | ||
| propertiesStart := 17 + message.RemarkLength |
There was a problem hiding this comment.
Same here about magical number 3,5,13,17
| "kyanos/agent/protocol" | ||
| ) | ||
|
|
||
| var _ protocol.ParsedMessage = &RocketMQMessage{} |
There was a problem hiding this comment.
Here you are validating the struct respect the interface.
This one and the previous one you do in the other files could be in tests file.
I'm unsure the compiler clean them in the final binary
There was a problem hiding this comment.
|
I know but I did it anyway 😅 |
|
@ccoVeille Thanks for the REVIEW, but currently in the stage of learning as I go along with the implementation 😅 |
Closes #189