From 1928ef5db02c9a8ed0e0fca8fa8f4529559d0502 Mon Sep 17 00:00:00 2001 From: yhsmer Date: Wed, 19 Apr 2023 17:45:25 +0800 Subject: [PATCH 1/6] fix --- .gitignore | 1 + .gitmodules | 4 +- .../docker/kindling-collector-config.yml | 7 ++- .../analyzer/network/message_pair.go | 27 +++++++---- .../analyzer/network/network_analyzer.go | 15 +++++- .../analyzer/network/protocol/protocol.go | 1 + collector/pkg/model/constnames/const.go | 7 ++- collector/pkg/model/kindling_event_helper.go | 4 +- probe/libs/agent-libs | 2 +- probe/src/cgo/kindling.cpp | 46 ++++++++++++++++++- probe/src/cgo/kindling.h | 5 +- 11 files changed, 99 insertions(+), 20 deletions(-) diff --git a/.gitignore b/.gitignore index 339a6a2eb..64f6bfe29 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ build/ probe/src/probe/cmake-build-debug/ probe/build/ probe/cmake-build-debug +probe/pkg collector/docker/kindling-collector collector/docker/libso/libkindling.so diff --git a/.gitmodules b/.gitmodules index 18ad8a026..5244f06ea 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "probe/libs/agent-libs"] path = probe/libs/agent-libs - url = https://github.com/KindlingProject/agent-libs.git - branch = kindling-dev + url = https://github.com/yhsmer/agent-libs.git + branch = feat/grpc diff --git a/collector/docker/kindling-collector-config.yml b/collector/docker/kindling-collector-config.yml index 7c630dd14..b446ef00e 100644 --- a/collector/docker/kindling-collector-config.yml +++ b/collector/docker/kindling-collector-config.yml @@ -33,6 +33,9 @@ receivers: - name: kretprobe-tcp_connect - name: kprobe-tcp_set_state - name: tracepoint-procexit + - name: uprobe-grpc_header_encoder + - name: uprobe-grpc_header_server_recv + - name: uprobe-grpc_header_client_recv process_filter: # the length of a comm should be no more than 16 comms: @@ -111,12 +114,14 @@ analyzers: - key: "rocketmq" ports: [ 9876, 10911 ] slow_threshold: 500 + - key: "grpc" + slow_threshold: 500 processors: k8smetadataprocessor: # Set "enable" false if you want to run the agent in the non-Kubernetes environment. # Otherwise, the agent will panic if it can't connect to the API-server. - enable: true + enable: false kube_auth_type: serviceAccount kube_config_dir: /root/.kube/config # GraceDeletePeriod controls the delay interval after receiving delete event. diff --git a/collector/pkg/component/analyzer/network/message_pair.go b/collector/pkg/component/analyzer/network/message_pair.go index 03e27d338..e8d8c4d26 100644 --- a/collector/pkg/component/analyzer/network/message_pair.go +++ b/collector/pkg/component/analyzer/network/message_pair.go @@ -354,12 +354,13 @@ func (mp *messagePair) getDuration() uint64 { // DNS will send different ip and port data with sharing fd and pid socket. type messagePairKey struct { - pid uint32 - fd int32 - sip string - dip string - sport uint32 - dport uint32 + pid uint32 + fd int32 + streamid uint32 + sip string + dip string + sport uint32 + dport uint32 } func getMessagePairKey(evt *model.KindlingEvent) messagePairKey { @@ -373,9 +374,17 @@ func getMessagePairKey(evt *model.KindlingEvent) messagePairKey { dport: evt.GetDport(), } } else { - return messagePairKey{ - pid: evt.GetPid(), - fd: evt.GetFd(), + if evt.GetUintUserAttribute("streamid") > 0 { + return messagePairKey{ + pid: evt.GetPid(), + fd: int32(evt.GetUintUserAttribute("fd")), + streamid: uint32(evt.GetUintUserAttribute("streamid")), + } + } else { + return messagePairKey{ + pid: evt.GetPid(), + fd: evt.GetFd(), + } } } } diff --git a/collector/pkg/component/analyzer/network/network_analyzer.go b/collector/pkg/component/analyzer/network/network_analyzer.go index adf91d7a0..65bce873e 100644 --- a/collector/pkg/component/analyzer/network/network_analyzer.go +++ b/collector/pkg/component/analyzer/network/network_analyzer.go @@ -101,6 +101,9 @@ func (na *NetworkAnalyzer) ConsumableEvents() []string { constnames.SendMsgEvent, constnames.RecvMsgEvent, constnames.SendMMsgEvent, + constnames.GrpcHeaderEncoder, + constnames.GrpcHeaderServerRecv, + constnames.GrpcHeaderClientRecv, } } @@ -186,7 +189,7 @@ func (na *NetworkAnalyzer) ConsumeEvent(evt *model.KindlingEvent) error { return na.analyseConnect(evt) } - if evt.GetDataLen() <= 0 || evt.GetResVal() < 0 { + if evt.GetUintUserAttribute("streamid") <= 0 && (evt.GetDataLen() <= 0 || evt.GetResVal() < 0) { // TODO: analyse udp return nil } @@ -319,6 +322,11 @@ func (na *NetworkAnalyzer) analyseResponse(evt *model.KindlingEvent) error { oldPairs.mergeResponse(evt) na.requestMonitor.Store(oldPairs.getKey(), oldPairs) + + if evt.GetUintUserAttribute("end_stream") == 1 { + _ = na.distributeTraceMetric(oldPairs, nil) + } + return nil } @@ -376,6 +384,11 @@ func (na *NetworkAnalyzer) distributeTraceMetric(oldPairs *messagePairs, newPair } func (na *NetworkAnalyzer) parseProtocols(mps *messagePairs) []*model.DataGroup { + // check grpc protocol + if mps.requests.event.GetUintUserAttribute("streamid") > 0 { + return na.getRecords(mps, protocol.GRPC, nil) + } + // Step 1: Static Config for port and protocol set in config file port := mps.getPort() staticProtocol, found := na.staticPortMap[port] diff --git a/collector/pkg/component/analyzer/network/protocol/protocol.go b/collector/pkg/component/analyzer/network/protocol/protocol.go index 823e25281..342777e34 100644 --- a/collector/pkg/component/analyzer/network/protocol/protocol.go +++ b/collector/pkg/component/analyzer/network/protocol/protocol.go @@ -8,6 +8,7 @@ const ( REDIS = "redis" DUBBO = "dubbo" ROCKETMQ = "rocketmq" + GRPC = "grpc" NOSUPPORT = "NOSUPPORT" ) diff --git a/collector/pkg/model/constnames/const.go b/collector/pkg/model/constnames/const.go index bb25d3f5a..e0877243f 100644 --- a/collector/pkg/model/constnames/const.go +++ b/collector/pkg/model/constnames/const.go @@ -29,8 +29,11 @@ const ( SpanEvent = "apm_span_event" OtherEvent = "other" - ProcessExitEvent = "procexit" - GrpcUprobeEvent = "grpc_uprobe" + ProcessExitEvent = "procexit" + GrpcUprobeEvent = "grpc_uprobe" + GrpcHeaderEncoder = "grpc_header_encoder" + GrpcHeaderServerRecv = "grpc_header_server_recv" + GrpcHeaderClientRecv = "grpc_header_client_recv" // NetRequestMetricGroupName is used for dataGroup generated from networkAnalyzer. NetRequestMetricGroupName = "net_request_metric_group" // SingleNetRequestMetricGroup stands for the dataGroup with abnormal status. diff --git a/collector/pkg/model/kindling_event_helper.go b/collector/pkg/model/kindling_event_helper.go index 18ab5d79c..95fd3b030 100644 --- a/collector/pkg/model/kindling_event_helper.go +++ b/collector/pkg/model/kindling_event_helper.go @@ -284,11 +284,11 @@ func (k *KindlingEvent) IsRequest() (bool, error) { switch k.Name { case constnames.ReadEvent, constnames.RecvFromEvent, constnames.RecvMsgEvent, constnames.ReadvEvent: fallthrough - case constnames.PReadEvent, constnames.PReadvEvent: + case constnames.PReadEvent, constnames.PReadvEvent, constnames.GrpcHeaderClientRecv, constnames.GrpcHeaderServerRecv: return k.isRequest(true) case constnames.WriteEvent, constnames.SendToEvent, constnames.SendMsgEvent, constnames.WritevEvent: fallthrough - case constnames.SendMMsgEvent, constnames.PWriteEvent, constnames.PWritevEvent: + case constnames.SendMMsgEvent, constnames.PWriteEvent, constnames.PWritevEvent, constnames.GrpcHeaderEncoder: return k.isRequest(false) default: break diff --git a/probe/libs/agent-libs b/probe/libs/agent-libs index 1daf5aa0e..96736b6a6 160000 --- a/probe/libs/agent-libs +++ b/probe/libs/agent-libs @@ -1 +1 @@ -Subproject commit 1daf5aa0e534cd96f9801729df35e5942c05f2bd +Subproject commit 96736b6a6624f6e2aee7b144ed9d1fa738fdc20e diff --git a/probe/src/cgo/kindling.cpp b/probe/src/cgo/kindling.cpp index 5c382b412..3e8f5cf0c 100644 --- a/probe/src/cgo/kindling.cpp +++ b/probe/src/cgo/kindling.cpp @@ -194,6 +194,43 @@ int getEvent(void** pp_kindling_event) { uint16_t ev_type = ev->get_type(); print_event(ev); + if(ev->get_type() == PPME_FUN_E){ + cout << ev->get_name() << ' ' << "parameter: " << *((uint32_t *)(ev->get_param_value_raw("parameter"))->m_val) << endl; + } + if(ev->get_type() == PPME_GRPC_HEADER_ENCODE_E){ + cout << ev->get_name() << " event ==> " << endl; + cout << "ts: " << ev->get_ts() << endl; + cout << "end_stream: " << *((uint32_t *)(ev->get_param_value_raw("end_stream"))->m_val) << endl; + cout << "streamid: " << *((uint32_t *)(ev->get_param_value_raw("streamid"))->m_val) << endl; + cout << "fd: " << *((int32_t *)(ev->get_param_value_raw("fd"))->m_val) << endl; + cout << "status: " << ((char *)(ev->get_param_value_raw("status"))->m_val) << endl; + cout << "grpc_status: " << ((char *)(ev->get_param_value_raw("grpc_status"))->m_val) << endl; + cout << "scheme: " << ((char *)(ev->get_param_value_raw("scheme"))->m_val) << endl; + cout << "authority: " << ((char *)(ev->get_param_value_raw("authority"))->m_val) << endl; + cout << "path: " << ((char *)(ev->get_param_value_raw("path"))->m_val) << endl; + cout << "latency: " << ev->get_thread_info()->m_latency << endl; + } + if(ev->get_type() == PPME_GRPC_HEADER_SERVER_RECV_E){ + cout << ev->get_name() << " event ==> " << endl; + cout << "ts: " << ev->get_ts() << endl; + cout << "end_stream: " << *((uint32_t *)(ev->get_param_value_raw("end_stream"))->m_val) << endl; + cout << "streamid: " << *((uint32_t *)(ev->get_param_value_raw("streamid"))->m_val) << endl; + cout << "fd: " << *((int32_t *)(ev->get_param_value_raw("fd"))->m_val) << endl; + cout << "scheme: " << ((char *)(ev->get_param_value_raw("scheme"))->m_val) << endl; + cout << "authority: " << ((char *)(ev->get_param_value_raw("authority"))->m_val) << endl; + cout << "path: " << ((char *)(ev->get_param_value_raw("path"))->m_val) << endl; + cout << "latency: " << ev->get_thread_info()->m_latency << endl; + } + if(ev->get_type() == PPME_GRPC_HEADER_CLIENT_RECV_E){ + cout << ev->get_name() << " event ==> " << endl; + cout << "ts: " << ev->get_ts() << endl; + cout << "end_stream: " << *((uint32_t *)(ev->get_param_value_raw("end_stream"))->m_val) << endl; + cout << "streamid: " << *((uint32_t *)(ev->get_param_value_raw("streamid"))->m_val) << endl; + cout << "fd: " << *((int32_t *)(ev->get_param_value_raw("fd"))->m_val) << endl; + cout << "status: " << ((char *)(ev->get_param_value_raw("status"))->m_val) << endl; + cout << "grpc_status: " << ((char *)(ev->get_param_value_raw("grpc_status"))->m_val) << endl; + cout << "latency: " << ev->get_thread_info()->m_latency << endl; + } if (ev_type != PPME_CPU_ANALYSIS_E && is_profile_debug && threadInfo->m_tid == debug_tid && threadInfo->m_pid == debug_pid) { print_profile_debug_info(ev); @@ -695,12 +732,15 @@ void init_kindling_event(kindling_event_t_for_go* p_kindling_event, void** pp_ki p_kindling_event->context.tinfo.containerId = (char*)malloc(sizeof(char) * 256); p_kindling_event->context.fdInfo.filename = (char*)malloc(sizeof(char) * 1024); p_kindling_event->context.fdInfo.directory = (char*)malloc(sizeof(char) * 1024); - for (int i = 0; i < 16; i++) { p_kindling_event->userAttributes[i].key = (char*)malloc(sizeof(char) * 128); p_kindling_event->userAttributes[i].value = (char*)malloc(sizeof(char) * EVENT_DATA_SIZE); } } + else{ + ((kindling_event_t_for_go*)*pp_kindling_event)->latency = 0; + } + } void print_event(sinsp_evt* s_evt) { @@ -898,6 +938,10 @@ uint16_t get_kindling_source(uint16_t etype) { case PPME_TCP_DROP_E: case PPME_TCP_RETRANCESMIT_SKB_E: return KRPOBE; + case PPME_GRPC_HEADER_ENCODE_E: + case PPME_GRPC_HEADER_SERVER_RECV_E: + case PPME_GRPC_HEADER_CLIENT_RECV_E: + return UPROBE; // TODO add cases of tracepoint, kprobe, uprobe default: return SYSCALL_ENTER; diff --git a/probe/src/cgo/kindling.h b/probe/src/cgo/kindling.h index e3dc66447..45cbfd968 100644 --- a/probe/src/cgo/kindling.h +++ b/probe/src/cgo/kindling.h @@ -7,7 +7,7 @@ #include #include -#include +#include KRPOBE #include #include #include @@ -435,6 +435,9 @@ const static event kindling_to_sysdig[PPM_EVENT_MAX] = { {"tracepoint-tcp_receive_reset", PPME_TCP_RECEIVE_RESET_E}, {"tracepoint-cpu_analysis", PPME_CPU_ANALYSIS_E}, {"tracepoint-procexit", PPME_PROCEXIT_1_E}, + {"uprobe-grpc_header_encoder", PPME_GRPC_HEADER_ENCODE_E}, + {"uprobe-grpc_header_server_recv", PPME_GRPC_HEADER_SERVER_RECV_E}, + {"uprobe-grpc_header_client_recv", PPME_GRPC_HEADER_CLIENT_RECV_E}, }; struct event_category { From 1bbc03912879e0a5c4bebce4d06eb4e1b85aefc3 Mon Sep 17 00:00:00 2001 From: yhsmer Date: Thu, 20 Apr 2023 18:59:58 +0800 Subject: [PATCH 2/6] achieve grpc protocol --- .../analyzer/network/message_pair.go | 7 ++++ .../analyzer/network/network_analyzer.go | 37 ++++++++++++++++++- collector/pkg/model/constlabels/const.go | 1 + collector/pkg/model/constlabels/protocols.go | 1 + 4 files changed, 45 insertions(+), 1 deletion(-) diff --git a/collector/pkg/component/analyzer/network/message_pair.go b/collector/pkg/component/analyzer/network/message_pair.go index e8d8c4d26..7e0c8f677 100644 --- a/collector/pkg/component/analyzer/network/message_pair.go +++ b/collector/pkg/component/analyzer/network/message_pair.go @@ -48,6 +48,13 @@ func (evts *events) getEvent(index int) *model.KindlingEvent { return nil } +func (evts *events) getEvents() []*model.KindlingEvent { + if evts.mergable != nil { + return evts.mergable.events + } + return []*model.KindlingEvent{evts.event} +} + func (evts *events) putEventBack(originEvts *events) { newEvt := evts.event evts.event = originEvts.event diff --git a/collector/pkg/component/analyzer/network/network_analyzer.go b/collector/pkg/component/analyzer/network/network_analyzer.go index 65bce873e..30ab3fcdf 100644 --- a/collector/pkg/component/analyzer/network/network_analyzer.go +++ b/collector/pkg/component/analyzer/network/network_analyzer.go @@ -5,6 +5,7 @@ import ( "math/rand" "os" "strconv" + "strings" "sync" "sync/atomic" "time" @@ -386,7 +387,7 @@ func (na *NetworkAnalyzer) distributeTraceMetric(oldPairs *messagePairs, newPair func (na *NetworkAnalyzer) parseProtocols(mps *messagePairs) []*model.DataGroup { // check grpc protocol if mps.requests.event.GetUintUserAttribute("streamid") > 0 { - return na.getRecords(mps, protocol.GRPC, nil) + return na.getRecords(mps, protocol.GRPC, generateGrpcAttributeMap(mps)) } // Step 1: Static Config for port and protocol set in config file @@ -742,3 +743,37 @@ func (na *NetworkAnalyzer) getResponseSlowThreshold(protocol string) int { } return na.cfg.getResponseSlowThreshold() } + +func generateGrpcAttributeMap(mps *messagePairs) *model.AttributeMap { + attributeMap := model.NewAttributeMap() + for _, event := range mps.requests.getEvents() { + attributeMap.AddStringValue("scheme", event.GetStringUserAttribute("scheme")) + attributeMap.AddStringValue("authority", event.GetStringUserAttribute("authority")) + attributeMap.AddStringValue("path", event.GetStringUserAttribute("path")) + } + for _, event := range mps.responses.getEvents() { + status := event.GetStringUserAttribute("status") + status = strings.ReplaceAll(status, "\x00", "") + + if status != "" { + statusCode, _ := strconv.ParseInt(status, 10, 64) + attributeMap.AddIntValue(constlabels.HttpStatusCode, statusCode) + if statusCode >= 400 { + attributeMap.AddBoolValue(constlabels.IsError, true) + attributeMap.AddIntValue(constlabels.ErrorType, int64(constlabels.ProtocolError)) + } + } + grpcStatus := event.GetStringUserAttribute("grpc_status") + grpcStatus = strings.ReplaceAll(grpcStatus, "\x00", "") + if grpcStatus != "" { + grpcStatusCode, _ := strconv.ParseInt(grpcStatus, 10, 64) + attributeMap.AddIntValue(constlabels.GrpcStatusCode, grpcStatusCode) + if grpcStatusCode > 0 { + attributeMap.AddBoolValue(constlabels.IsError, true) + attributeMap.AddIntValue(constlabels.ErrorType, int64(constlabels.GrpcError)) + } + } + + } + return attributeMap +} diff --git a/collector/pkg/model/constlabels/const.go b/collector/pkg/model/constlabels/const.go index ed397645e..77b7d8c07 100644 --- a/collector/pkg/model/constlabels/const.go +++ b/collector/pkg/model/constlabels/const.go @@ -5,6 +5,7 @@ const ( ConnectFail NoResponse ProtocolError + GrpcError ) const ( diff --git a/collector/pkg/model/constlabels/protocols.go b/collector/pkg/model/constlabels/protocols.go index 06fb29e4f..9a157e8ac 100644 --- a/collector/pkg/model/constlabels/protocols.go +++ b/collector/pkg/model/constlabels/protocols.go @@ -10,6 +10,7 @@ const ( HttpApmTraceType = "trace_type" HttpApmTraceId = "trace_id" HttpStatusCode = "http_status_code" + GrpcStatusCode = "grpc_status_code" HttpContinue = "http_continue" DnsId = "dns_id" From 911c6f7f112169f59e1ccfb9e6dfd509d901276d Mon Sep 17 00:00:00 2001 From: yhsmer Date: Tue, 25 Apr 2023 12:47:39 +0800 Subject: [PATCH 3/6] achieve grpc protocol analysis --- .gitmodules | 4 +- collector/pkg/aggregator/label_key.go | 2 +- .../analyzer/network/message_pair.go | 7 --- .../analyzer/network/network_analyzer.go | 37 ++++++++++---- .../processor/aggregateprocessor/processor.go | 7 +++ collector/pkg/model/constlabels/protocols.go | 7 ++- deploy/agent/kindling-collector-config.yml | 5 ++ deploy/agent/kindling-deploy.yml | 48 ++----------------- probe/src/cgo/kindling.cpp | 37 -------------- 9 files changed, 53 insertions(+), 101 deletions(-) diff --git a/.gitmodules b/.gitmodules index 5244f06ea..e701629ab 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "probe/libs/agent-libs"] path = probe/libs/agent-libs - url = https://github.com/yhsmer/agent-libs.git - branch = feat/grpc + url = https://github.com/KindlingProject/agent-libs.git + branch = kindling-dev \ No newline at end of file diff --git a/collector/pkg/aggregator/label_key.go b/collector/pkg/aggregator/label_key.go index 9cd25a08a..dc5912c57 100644 --- a/collector/pkg/aggregator/label_key.go +++ b/collector/pkg/aggregator/label_key.go @@ -50,7 +50,7 @@ func (s *LabelSelectors) AppendSelectors(selectors ...LabelSelector) { s.selectors = append(s.selectors, selectors...) } -const maxLabelKeySize = 37 +const maxLabelKeySize = 41 type LabelKeys struct { // LabelKeys will be used as key of map, so it is must be an array instead of a slice. diff --git a/collector/pkg/component/analyzer/network/message_pair.go b/collector/pkg/component/analyzer/network/message_pair.go index 7e0c8f677..e8d8c4d26 100644 --- a/collector/pkg/component/analyzer/network/message_pair.go +++ b/collector/pkg/component/analyzer/network/message_pair.go @@ -48,13 +48,6 @@ func (evts *events) getEvent(index int) *model.KindlingEvent { return nil } -func (evts *events) getEvents() []*model.KindlingEvent { - if evts.mergable != nil { - return evts.mergable.events - } - return []*model.KindlingEvent{evts.event} -} - func (evts *events) putEventBack(originEvts *events) { newEvt := evts.event evts.event = originEvts.event diff --git a/collector/pkg/component/analyzer/network/network_analyzer.go b/collector/pkg/component/analyzer/network/network_analyzer.go index 30ab3fcdf..bf44eeda0 100644 --- a/collector/pkg/component/analyzer/network/network_analyzer.go +++ b/collector/pkg/component/analyzer/network/network_analyzer.go @@ -634,6 +634,12 @@ func (na *NetworkAnalyzer) getRecords(mps *messagePairs, protocol string, attrib ret.UpdateAddIntMetric(constvalues.RequestIo, int64(mps.getRquestSize())) ret.UpdateAddIntMetric(constvalues.ResponseIo, int64(mps.getResponseSize())) + //TODO: get grpc frame data size, then update these value + if evt.GetUintUserAttribute("streamid") != 0 { + ret.UpdateAddIntMetric(constvalues.RequestIo, 0) + ret.UpdateAddIntMetric(constvalues.ResponseIo, 0) + } + ret.Timestamp = evt.GetStartTime() return []*model.DataGroup{ret} @@ -746,15 +752,25 @@ func (na *NetworkAnalyzer) getResponseSlowThreshold(protocol string) int { func generateGrpcAttributeMap(mps *messagePairs) *model.AttributeMap { attributeMap := model.NewAttributeMap() - for _, event := range mps.requests.getEvents() { - attributeMap.AddStringValue("scheme", event.GetStringUserAttribute("scheme")) - attributeMap.AddStringValue("authority", event.GetStringUserAttribute("authority")) - attributeMap.AddStringValue("path", event.GetStringUserAttribute("path")) + if mps == nil || mps.requests == nil { + return attributeMap + } + + request := mps.requests.getEvent(0) + if request != nil { + attributeMap.AddStringValue(constlabels.Scheme, strings.ReplaceAll(request.GetStringUserAttribute("scheme"), "\x00", "")) + attributeMap.AddStringValue(constlabels.Authority, strings.ReplaceAll(request.GetStringUserAttribute("authority"), "\x00", "")) + attributeMap.AddStringValue(constlabels.Path, strings.ReplaceAll(request.GetStringUserAttribute("path"), "\x00", "")) + } + + if mps.responses == nil { + return attributeMap } - for _, event := range mps.responses.getEvents() { - status := event.GetStringUserAttribute("status") - status = strings.ReplaceAll(status, "\x00", "") + firstResp := mps.responses.getEvent(0) + if firstResp != nil { + status := firstResp.GetStringUserAttribute("status") + status = strings.ReplaceAll(status, "\x00", "") if status != "" { statusCode, _ := strconv.ParseInt(status, 10, 64) attributeMap.AddIntValue(constlabels.HttpStatusCode, statusCode) @@ -763,7 +779,11 @@ func generateGrpcAttributeMap(mps *messagePairs) *model.AttributeMap { attributeMap.AddIntValue(constlabels.ErrorType, int64(constlabels.ProtocolError)) } } - grpcStatus := event.GetStringUserAttribute("grpc_status") + } + + lastResp := mps.responses.getEvent(mps.responses.size() - 1) + if lastResp != nil { + grpcStatus := lastResp.GetStringUserAttribute("grpc_status") grpcStatus = strings.ReplaceAll(grpcStatus, "\x00", "") if grpcStatus != "" { grpcStatusCode, _ := strconv.ParseInt(grpcStatus, 10, 64) @@ -773,7 +793,6 @@ func generateGrpcAttributeMap(mps *messagePairs) *model.AttributeMap { attributeMap.AddIntValue(constlabels.ErrorType, int64(constlabels.GrpcError)) } } - } return attributeMap } diff --git a/collector/pkg/component/consumer/processor/aggregateprocessor/processor.go b/collector/pkg/component/consumer/processor/aggregateprocessor/processor.go index 349fbb32c..d06df90df 100644 --- a/collector/pkg/component/consumer/processor/aggregateprocessor/processor.go +++ b/collector/pkg/component/consumer/processor/aggregateprocessor/processor.go @@ -182,6 +182,13 @@ func newNetRequestLabelSelectors() *aggregator.LabelSelectors { aggregator.LabelSelector{Name: constlabels.DnsDomain, VType: aggregator.StringType}, aggregator.LabelSelector{Name: constlabels.KafkaTopic, VType: aggregator.StringType}, aggregator.LabelSelector{Name: constlabels.RocketMQErrCode, VType: aggregator.IntType}, + + // grpc request scheme authority path + aggregator.LabelSelector{Name: constlabels.Scheme, VType: aggregator.StringType}, + aggregator.LabelSelector{Name: constlabels.Authority, VType: aggregator.StringType}, + aggregator.LabelSelector{Name: constlabels.Path, VType: aggregator.StringType}, + // rpc status code + aggregator.LabelSelector{Name: constlabels.GrpcStatusCode, VType: aggregator.IntType}, ) } diff --git a/collector/pkg/model/constlabels/protocols.go b/collector/pkg/model/constlabels/protocols.go index 9a157e8ac..05bbb2f36 100644 --- a/collector/pkg/model/constlabels/protocols.go +++ b/collector/pkg/model/constlabels/protocols.go @@ -10,9 +10,14 @@ const ( HttpApmTraceType = "trace_type" HttpApmTraceId = "trace_id" HttpStatusCode = "http_status_code" - GrpcStatusCode = "grpc_status_code" HttpContinue = "http_continue" + GrpcStatusCode = "grpc_status_code" + Scheme = "scheme" + Authority = "authority" + Path = "path" + StreamId = "stream_id" + DnsId = "dns_id" DnsDomain = "dns_domain" DnsRcode = "dns_rcode" diff --git a/deploy/agent/kindling-collector-config.yml b/deploy/agent/kindling-collector-config.yml index 7c630dd14..09aeba5cd 100644 --- a/deploy/agent/kindling-collector-config.yml +++ b/deploy/agent/kindling-collector-config.yml @@ -33,6 +33,9 @@ receivers: - name: kretprobe-tcp_connect - name: kprobe-tcp_set_state - name: tracepoint-procexit + - name: uprobe-grpc_header_encoder + - name: uprobe-grpc_header_server_recv + - name: uprobe-grpc_header_client_recv process_filter: # the length of a comm should be no more than 16 comms: @@ -111,6 +114,8 @@ analyzers: - key: "rocketmq" ports: [ 9876, 10911 ] slow_threshold: 500 + - key: "grpc" + slow_threshold: 500 processors: k8smetadataprocessor: diff --git a/deploy/agent/kindling-deploy.yml b/deploy/agent/kindling-deploy.yml index 79ef775c4..0e1a4bba1 100644 --- a/deploy/agent/kindling-deploy.yml +++ b/deploy/agent/kindling-deploy.yml @@ -82,27 +82,8 @@ spec: - mountPath: /etc/modprobe.d name: modprobe-d readOnly: true - - mountPath: /host/dev - name: dev-vol - - mountPath: /host/proc - name: proc-vol - readOnly: true - - mountPath: /host/etc - name: etc-vol - readOnly: true - - mountPath: /host/boot - name: boot-vol - readOnly: true - - mountPath: /host/lib/modules - name: modules-vol - readOnly: true - - mountPath: /host/usr - name: usr-vol - readOnly: true - - mountPath: /host/run - name: run-vol - - mountPath: /host/var/run - name: varrun-vol + - mountPath: /host/ + name: root-vol - mountPath: /dev/shm name: dshm dnsPolicy: ClusterFirstWithHostNet @@ -128,29 +109,8 @@ spec: medium: Memory name: dshm - hostPath: - path: /dev - name: dev-vol - - hostPath: - path: /proc - name: proc-vol - - hostPath: - path: /etc - name: etc-vol - - hostPath: - path: /boot - name: boot-vol - - hostPath: - path: /lib/modules - name: modules-vol - - hostPath: - path: /usr - name: usr-vol - - hostPath: - path: /run - name: run-vol - - hostPath: - path: /var/run - name: varrun-vol + path: / + name: root-vol - hostPath: path: /sys name: sys-vol \ No newline at end of file diff --git a/probe/src/cgo/kindling.cpp b/probe/src/cgo/kindling.cpp index 3e8f5cf0c..a159fe1db 100644 --- a/probe/src/cgo/kindling.cpp +++ b/probe/src/cgo/kindling.cpp @@ -194,43 +194,6 @@ int getEvent(void** pp_kindling_event) { uint16_t ev_type = ev->get_type(); print_event(ev); - if(ev->get_type() == PPME_FUN_E){ - cout << ev->get_name() << ' ' << "parameter: " << *((uint32_t *)(ev->get_param_value_raw("parameter"))->m_val) << endl; - } - if(ev->get_type() == PPME_GRPC_HEADER_ENCODE_E){ - cout << ev->get_name() << " event ==> " << endl; - cout << "ts: " << ev->get_ts() << endl; - cout << "end_stream: " << *((uint32_t *)(ev->get_param_value_raw("end_stream"))->m_val) << endl; - cout << "streamid: " << *((uint32_t *)(ev->get_param_value_raw("streamid"))->m_val) << endl; - cout << "fd: " << *((int32_t *)(ev->get_param_value_raw("fd"))->m_val) << endl; - cout << "status: " << ((char *)(ev->get_param_value_raw("status"))->m_val) << endl; - cout << "grpc_status: " << ((char *)(ev->get_param_value_raw("grpc_status"))->m_val) << endl; - cout << "scheme: " << ((char *)(ev->get_param_value_raw("scheme"))->m_val) << endl; - cout << "authority: " << ((char *)(ev->get_param_value_raw("authority"))->m_val) << endl; - cout << "path: " << ((char *)(ev->get_param_value_raw("path"))->m_val) << endl; - cout << "latency: " << ev->get_thread_info()->m_latency << endl; - } - if(ev->get_type() == PPME_GRPC_HEADER_SERVER_RECV_E){ - cout << ev->get_name() << " event ==> " << endl; - cout << "ts: " << ev->get_ts() << endl; - cout << "end_stream: " << *((uint32_t *)(ev->get_param_value_raw("end_stream"))->m_val) << endl; - cout << "streamid: " << *((uint32_t *)(ev->get_param_value_raw("streamid"))->m_val) << endl; - cout << "fd: " << *((int32_t *)(ev->get_param_value_raw("fd"))->m_val) << endl; - cout << "scheme: " << ((char *)(ev->get_param_value_raw("scheme"))->m_val) << endl; - cout << "authority: " << ((char *)(ev->get_param_value_raw("authority"))->m_val) << endl; - cout << "path: " << ((char *)(ev->get_param_value_raw("path"))->m_val) << endl; - cout << "latency: " << ev->get_thread_info()->m_latency << endl; - } - if(ev->get_type() == PPME_GRPC_HEADER_CLIENT_RECV_E){ - cout << ev->get_name() << " event ==> " << endl; - cout << "ts: " << ev->get_ts() << endl; - cout << "end_stream: " << *((uint32_t *)(ev->get_param_value_raw("end_stream"))->m_val) << endl; - cout << "streamid: " << *((uint32_t *)(ev->get_param_value_raw("streamid"))->m_val) << endl; - cout << "fd: " << *((int32_t *)(ev->get_param_value_raw("fd"))->m_val) << endl; - cout << "status: " << ((char *)(ev->get_param_value_raw("status"))->m_val) << endl; - cout << "grpc_status: " << ((char *)(ev->get_param_value_raw("grpc_status"))->m_val) << endl; - cout << "latency: " << ev->get_thread_info()->m_latency << endl; - } if (ev_type != PPME_CPU_ANALYSIS_E && is_profile_debug && threadInfo->m_tid == debug_tid && threadInfo->m_pid == debug_pid) { print_profile_debug_info(ev); From c1f0c734f323fb3473c8d0a12e4e34552f4fa5e5 Mon Sep 17 00:00:00 2001 From: yhsmer Date: Tue, 25 Apr 2023 13:02:37 +0800 Subject: [PATCH 4/6] update submodule --- probe/libs/agent-libs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/probe/libs/agent-libs b/probe/libs/agent-libs index 96736b6a6..1daf5aa0e 160000 --- a/probe/libs/agent-libs +++ b/probe/libs/agent-libs @@ -1 +1 @@ -Subproject commit 96736b6a6624f6e2aee7b144ed9d1fa738fdc20e +Subproject commit 1daf5aa0e534cd96f9801729df35e5942c05f2bd From 67e6c9c2b43d4197b09b0ffdd4e71757882745c6 Mon Sep 17 00:00:00 2001 From: yhsmer Date: Tue, 25 Apr 2023 13:36:26 +0800 Subject: [PATCH 5/6] do some clean --- .gitignore | 3 +-- collector/docker/kindling-collector-config.yml | 4 ++-- probe/src/cgo/kindling.h | 2 +- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/.gitignore b/.gitignore index 64f6bfe29..24c2f9a4a 100644 --- a/.gitignore +++ b/.gitignore @@ -5,7 +5,6 @@ build/ probe/src/probe/cmake-build-debug/ probe/build/ probe/cmake-build-debug -probe/pkg collector/docker/kindling-collector collector/docker/libso/libkindling.so @@ -13,4 +12,4 @@ collector/docker/kindling-falcolib-probe.tar.gz collector/vendor/ node_modules *.log -logs \ No newline at end of file +logs diff --git a/collector/docker/kindling-collector-config.yml b/collector/docker/kindling-collector-config.yml index a6b58d44e..b3eb6f754 100644 --- a/collector/docker/kindling-collector-config.yml +++ b/collector/docker/kindling-collector-config.yml @@ -121,7 +121,7 @@ processors: k8smetadataprocessor: # Set "enable" false if you want to run the agent in the non-Kubernetes environment. # Otherwise, the agent will panic if it can't connect to the API-server. - enable: false + enable: true kube_auth_type: serviceAccount kube_config_dir: /root/.kube/config # GraceDeletePeriod controls the delay interval after receiving delete event. @@ -247,4 +247,4 @@ observability: # Note: DO NOT add the prefix "http://" endpoint: 10.10.10.10:8080 stdout: - collect_period: 15s \ No newline at end of file + collect_period: 15s diff --git a/probe/src/cgo/kindling.h b/probe/src/cgo/kindling.h index 45cbfd968..580592a09 100644 --- a/probe/src/cgo/kindling.h +++ b/probe/src/cgo/kindling.h @@ -7,7 +7,7 @@ #include #include -#include KRPOBE +#include #include #include #include From 426a15487eb5e03faeacddbb69475d570842a2ce Mon Sep 17 00:00:00 2001 From: yhsmer Date: Thu, 4 May 2023 11:30:34 +0800 Subject: [PATCH 6/6] use env to enable uprobe and default is unenabled --- deploy/agent/kindling-deploy.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/deploy/agent/kindling-deploy.yml b/deploy/agent/kindling-deploy.yml index 0e1a4bba1..bfdbd1a13 100644 --- a/deploy/agent/kindling-deploy.yml +++ b/deploy/agent/kindling-deploy.yml @@ -59,6 +59,8 @@ spec: value: "200" - name: switch_agg_num value: "2" + - name: enable_uprobe + value: "false" - name: MY_NODE_IP valueFrom: fieldRef: