diff --git a/.gitignore b/.gitignore index 339a6a2eb..24c2f9a4a 100644 --- a/.gitignore +++ b/.gitignore @@ -12,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/.gitmodules b/.gitmodules index 18ad8a026..e701629ab 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 + branch = kindling-dev \ No newline at end of file diff --git a/collector/docker/kindling-collector-config.yml b/collector/docker/kindling-collector-config.yml index 8062dda93..b3eb6f754 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,6 +114,8 @@ analyzers: - key: "rocketmq" ports: [ 9876, 10911 ] slow_threshold: 500 + - key: "grpc" + slow_threshold: 500 processors: k8smetadataprocessor: @@ -242,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/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 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..bf44eeda0 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" @@ -101,6 +102,9 @@ func (na *NetworkAnalyzer) ConsumableEvents() []string { constnames.SendMsgEvent, constnames.RecvMsgEvent, constnames.SendMMsgEvent, + constnames.GrpcHeaderEncoder, + constnames.GrpcHeaderServerRecv, + constnames.GrpcHeaderClientRecv, } } @@ -186,7 +190,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 +323,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 +385,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, generateGrpcAttributeMap(mps)) + } + // Step 1: Static Config for port and protocol set in config file port := mps.getPort() staticProtocol, found := na.staticPortMap[port] @@ -620,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} @@ -729,3 +749,50 @@ func (na *NetworkAnalyzer) getResponseSlowThreshold(protocol string) int { } return na.cfg.getResponseSlowThreshold() } + +func generateGrpcAttributeMap(mps *messagePairs) *model.AttributeMap { + attributeMap := model.NewAttributeMap() + 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 + } + + 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) + if statusCode >= 400 { + attributeMap.AddBoolValue(constlabels.IsError, true) + attributeMap.AddIntValue(constlabels.ErrorType, int64(constlabels.ProtocolError)) + } + } + } + + 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) + 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/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/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/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..05bbb2f36 100644 --- a/collector/pkg/model/constlabels/protocols.go +++ b/collector/pkg/model/constlabels/protocols.go @@ -12,6 +12,12 @@ const ( HttpStatusCode = "http_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/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/deploy/agent/kindling-collector-config.yml b/deploy/agent/kindling-collector-config.yml index 8062dda93..9411dd66e 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..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: @@ -82,27 +84,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 +111,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 5c382b412..a159fe1db 100644 --- a/probe/src/cgo/kindling.cpp +++ b/probe/src/cgo/kindling.cpp @@ -695,12 +695,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 +901,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..580592a09 100644 --- a/probe/src/cgo/kindling.h +++ b/probe/src/cgo/kindling.h @@ -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 {