Merge pull request #243 from coroot/grpc_status

grpc: get grpc-status from payload if possible

Author: def

containers/container.go

@@ -741,8 +741,12 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R
 		requests := conn.http2Parser.Parse(r.Method, r.Payload, uint64(r.Duration))
 		for _, req := range requests {
 			if !common.HttpFilter.ShouldBeSkipped(req.Path) {
-				stats.observe(req.Status.Http(), "", req.Duration)
-				trace.Http2Request(req.Method, req.Path, req.Scheme, req.Status, req.Duration)
+				status := req.Status.Http()
+				if req.GrpcStatus >= 0 {
+					status = req.GrpcStatus.GRPC()
+				}
+				stats.observe(status, "", req.Duration)
+				trace.Http2Request(req.Method, req.Path, req.Scheme, req.Status, req.GrpcStatus, req.Duration)
 			}
 		}
 	case l7.ProtocolPostgres:

ebpftracer/l7/http2.go

@@ -24,11 +24,12 @@ type Http2FrameHeader struct {
 }
 
 type Http2Request struct {
-	Method   string
-	Path     string
-	Scheme   string
-	Status   Status
-	Duration time.Duration
+	Method     string
+	Path       string
+	Scheme     string
+	Status     Status
+	GrpcStatus Status
+	Duration   time.Duration
 
 	kernelTime uint64
 }
@@ -61,6 +62,8 @@ func (p *Http2Parser) Parse(method Method, payload []byte, kernelTime uint64) []
 
 	var decoder *hpack.Decoder
 	statuses := map[uint32]Status{}
+	grpcStatuses := map[uint32]Status{}
+
 	offset := 0
 
 	switch method {
@@ -119,9 +122,13 @@ func (p *Http2Parser) Parse(method Method, payload []byte, kernelTime uint64) []
 				statuses[h.StreamId] = 0
 			}
 			decoder.SetEmitFunc(func(hf hpack.HeaderField) {
-				if hf.Name == ":status" {
+				switch hf.Name {
+				case ":status":
 					s, _ := strconv.Atoi(hf.Value)
 					statuses[h.StreamId] = Status(s)
+				case "grpc-status":
+					s, _ := strconv.Atoi(hf.Value)
+					grpcStatuses[h.StreamId] = Status(s)
 				}
 			})
 		}
@@ -141,6 +148,12 @@ func (p *Http2Parser) Parse(method Method, payload []byte, kernelTime uint64) []
 			continue
 		}
 		r.Status = status
+		grpcStatus, ok := grpcStatuses[streamId]
+		if ok {
+			r.GrpcStatus = grpcStatus
+		} else {
+			r.GrpcStatus = -1
+		}
 		r.Duration = time.Duration(kernelTime - r.kernelTime)
 		res = append(res, *r)
 		delete(p.activeRequests, streamId)

ebpftracer/l7/l7.go

@@ -160,6 +160,46 @@ func (s Status) Zookeeper() string {
 	return "ok"
 }
 
+func (s Status) GRPC() string {
+	switch s {
+	case 0:
+		return "grpc:OK"
+	case 1:
+		return "grpc:CANCELLED"
+	case 2:
+		return "grpc:UNKNOWN"
+	case 3:
+		return "grpc:INVALID_ARGUMENT"
+	case 4:
+		return "grpc:DEADLINE_EXCEEDED"
+	case 5:
+		return "grpc:NOT_FOUND"
+	case 6:
+		return "grpc:ALREADY_EXISTS"
+	case 7:
+		return "grpc:PERMISSION_DENIED"
+	case 8:
+		return "grpc:RESOURCE_EXHAUSTED"
+	case 9:
+		return "grpc:FAILED_PRECONDITION"
+	case 10:
+		return "grpc:ABORTED"
+	case 11:
+		return "grpc:OUT_OF_RANGE"
+	case 12:
+		return "grpc:UNIMPLEMENTED"
+	case 13:
+		return "grpc:INTERNAL"
+	case 14:
+		return "grpc:UNAVAILABLE"
+	case 15:
+		return "grpc:DATA_LOSS"
+	case 16:
+		return "grpc:UNAUTHENTICATED"
+	}
+	return ""
+}
+
 func (s Status) Error() bool {
 	return s == StatusFailed
 }

tracing/tracing.go

@@ -151,7 +151,7 @@ func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time
 	)
 }
 
-func (t *Trace) Http2Request(method, path, scheme string, status l7.Status, duration time.Duration) {
+func (t *Trace) Http2Request(method, path, scheme string, status, grpcStatus l7.Status, duration time.Duration) {
 	if t == nil {
 		return
 	}
@@ -164,11 +164,16 @@ func (t *Trace) Http2Request(method, path, scheme string, status l7.Status, dura
 	if scheme == "" {
 		scheme = "unknown"
 	}
-	t.createSpan(method, duration, status > 400,
+
+	attrs := []attribute.KeyValue{
 		semconv.HTTPURL(fmt.Sprintf("%s://%s%s", scheme, t.destination.String(), path)),
 		semconv.HTTPMethod(method),
 		semconv.HTTPStatusCode(int(status)),
-	)
+	}
+	if grpcStatus >= 0 {
+		attrs = append(attrs, semconv.RPCGRPCStatusCodeKey.Int(int(grpcStatus)))
+	}
+	t.createSpan(method, duration, status > 400 || grpcStatus > 0, attrs...)
 }
 
 func (t *Trace) PostgresQuery(query string, error bool, duration time.Duration) {