achieve grpc protocol analysis

.gitignore

@@ -12,4 +12,4 @@ collector/docker/kindling-falcolib-probe.tar.gz
 collector/vendor/
 node_modules
 *.log
-logs
+logs

.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

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
+      collect_period: 15s

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.

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(),
+			}
 		}
 	}
 }

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
+}

collector/pkg/component/analyzer/network/protocol/protocol.go

@@ -8,6 +8,7 @@ const (
 	REDIS     = "redis"
 	DUBBO     = "dubbo"
 	ROCKETMQ  = "rocketmq"
+	GRPC      = "grpc"
 	NOSUPPORT = "NOSUPPORT"
 )
 

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},
 	)
 }
 

collector/pkg/model/constlabels/const.go

@@ -5,6 +5,7 @@ const (
 	ConnectFail
 	NoResponse
 	ProtocolError
+	GrpcError
 )
 
 const (

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"

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.

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

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:

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

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;