Instrument gRPC server span status codes

CHANGELOG.md

@@ -33,6 +33,7 @@ OpenTelemetry Go Automatic Instrumentation adheres to [Semantic Versioning](http
   An `slog.Logger` can now be configured by the user any way they want and then passed to the `Instrumentation` for its logging with this option. ([#1080](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1080))
 - Support `google.golang.org/grpc` `1.66.2`. ([#1083](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1083))
 - Support `google.golang.org/grpc` `1.67.0`. ([#1116](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1116))
+- Add gRPC status code attribute for server spans (`rpc.grpc.status_code`). ([#1127](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1127))
 
 ### Changed
 

go.mod

@@ -30,6 +30,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.30.0
 	golang.org/x/arch v0.10.0
 	golang.org/x/sys v0.25.0
+	google.golang.org/grpc v1.66.1
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -75,6 +76,5 @@ require (
 	golang.org/x/text v0.18.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
-	google.golang.org/grpc v1.66.1 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
 )

internal/pkg/instrumentation/bpf/google.golang.org/grpc/server/bpf/probe.bpf.c

@@ -19,6 +19,7 @@ struct grpc_request_t
 {
     BASE_SPAN_PROPERTIES
     char method[MAX_SIZE];
+    u32 status_code;
 };
 
 struct
@@ -59,6 +60,8 @@ volatile const u64 frame_stream_id_pod;
 volatile const u64 stream_id_pos;
 volatile const u64 stream_ctx_pos;
 volatile const bool is_new_frame_pos;
+volatile const u64 status_s_pos;
+volatile const u64 status_code_pos;
 
 static __always_inline long dummy_extract_span_context_from_headers(void *stream_id, struct span_context *parent_span_context) {
     return 0;
@@ -113,14 +116,14 @@ int uprobe_server_handleStream(struct pt_regs *ctx)
     if (!get_go_string_from_user_ptr((void *)(stream_ptr + stream_method_ptr_pos), grpcReq->method, sizeof(grpcReq->method)))
     {
         bpf_printk("Failed to read gRPC method from stream");
-        goto done;
+        bpf_map_delete_elem(&streamid_to_grpc_events, &stream_id);
+        return 0;
     }
 
     // Write event
     bpf_map_update_elem(&grpc_events, &key, grpcReq, 0);
     start_tracking_span(go_context.data, &grpcReq->sc);
-done:
-    bpf_map_delete_elem(&streamid_to_grpc_events, &stream_id);
+
     return 0;
 }
 
@@ -167,3 +170,48 @@ int uprobe_http2Server_operateHeader(struct pt_regs *ctx)
 
     return 0;
 }
+
+static __always_inline int get_status_code(struct pt_regs *ctx) {
+    struct go_iface go_context = {0};
+    get_Go_context(ctx, 2, stream_ctx_pos, true, &go_context);
+    void *key = get_consistent_key(ctx, go_context.data);
+
+    // Get parent context if exists
+    void *stream_ptr = get_argument(ctx, 2);
+    u32 stream_id = 0;
+    bpf_probe_read(&stream_id, sizeof(stream_id), (void *)(stream_ptr + stream_id_pos));
+    struct grpc_request_t *grpcReq = bpf_map_lookup_elem(&streamid_to_grpc_events, &stream_id);
+    if (grpcReq == NULL) {
+        // No parent span context, generate new span context
+        u32 zero = 0;
+        grpcReq = bpf_map_lookup_elem(&grpc_storage_map, &zero);
+        if (grpcReq == NULL) {
+            bpf_printk("failed to get grpcReq from storage map");
+            return 0;
+        }
+    }
+
+    void *status_ptr = get_argument(ctx, 3);
+    void *s_ptr = 0;
+    bpf_probe_read_user(&s_ptr, sizeof(s_ptr), (void *)(status_ptr + status_s_pos));
+    // Get status code from Status.s pointer
+    bpf_probe_read_user(&grpcReq->status_code, sizeof(grpcReq->status_code), (void *)(s_ptr + status_code_pos));
+
+    bpf_map_update_elem(&grpc_events, &key, grpcReq, 0);
+    bpf_map_delete_elem(&streamid_to_grpc_events, &stream_id);
+    return 0;
+}
+
+// func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status)
+// https://github.com/grpc/grpc-go/blob/bcf9171a20e44ed81a6eb152e3ca9e35b2c02c5d/internal/transport/handler_server.go#L228
+SEC("uprobe/serverHandlerTransport_WriteStatus")
+int uprobe_serverHandlerTransport_WriteStatus(struct pt_regs *ctx) {
+    return get_status_code(ctx);
+}
+
+// func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status)
+// https://github.com/grpc/grpc-go/blob/bcf9171a20e44ed81a6eb152e3ca9e35b2c02c5d/internal/transport/http2_server.go#L1049
+SEC("uprobe/http2Server_WriteStatus")
+int uprobe_http2Server_WriteStatus(struct pt_regs *ctx) {
+    return get_status_code(ctx);
+}

internal/pkg/instrumentation/bpf/google.golang.org/grpc/server/probe.go

@@ -8,10 +8,13 @@ import (
 	"log/slog"
 
 	"github.com/hashicorp/go-version"
+	"golang.org/x/sys/unix"
+	"google.golang.org/grpc/codes"
+
 	"go.opentelemetry.io/otel/attribute"
+	otelcodes "go.opentelemetry.io/otel/codes"
 	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 	"go.opentelemetry.io/otel/trace"
-	"golang.org/x/sys/unix"
 
 	"go.opentelemetry.io/auto/internal/pkg/inject"
 	"go.opentelemetry.io/auto/internal/pkg/instrumentation/context"
@@ -60,6 +63,14 @@ func New(logger *slog.Logger) probe.Probe {
 				Key: "frame_stream_id_pod",
 				Val: structfield.NewID("golang.org/x/net", "golang.org/x/net/http2", "FrameHeader", "StreamID"),
 			},
+			probe.StructFieldConst{
+				Key: "status_s_pos",
+				Val: structfield.NewID("google.golang.org/grpc", "google.golang.org/grpc/internal/status", "Status", "s"),
+			},
+			probe.StructFieldConst{
+				Key: "status_code_pos",
+				Val: structfield.NewID("google.golang.org/grpc", "google.golang.org/genproto/googleapis/rpc/status", "Status", "Code"),
+			},
 			framePosConst{},
 		},
 		Uprobes: []probe.Uprobe{
@@ -72,6 +83,14 @@ func New(logger *slog.Logger) probe.Probe {
 				Sym:        "google.golang.org/grpc/internal/transport.(*http2Server).operateHeaders",
 				EntryProbe: "uprobe_http2Server_operateHeader",
 			},
+			{
+				Sym:        "google.golang.org/grpc/internal/transport.(*serverHandlerTransport).WriteStatus",
+				EntryProbe: "uprobe_serverHandlerTransport_WriteStatus",
+			},
+			{
+				Sym:        "google.golang.org/grpc/internal/transport.(*http2Server).WriteStatus",
+				EntryProbe: "uprobe_http2Server_WriteStatus",
+			},
 		},
 		SpecFn:    loadBpf,
 		ProcessFn: convertEvent,
@@ -100,7 +119,8 @@ func (c framePosConst) InjectOption(td *process.TargetDetails) (inject.Option, e
 // event represents an event in the gRPC server during a gRPC request.
 type event struct {
 	context.BaseSpanProperties
-	Method [100]byte
+	Method     [100]byte
+	StatusCode int32
 }
 
 func convertEvent(e *event) []*probe.SpanEvent {
@@ -125,18 +145,29 @@ func convertEvent(e *event) []*probe.SpanEvent {
 		pscPtr = nil
 	}
 
-	return []*probe.SpanEvent{
-		{
-			SpanName:  method,
-			StartTime: utils.BootOffsetToTime(e.StartTime),
-			EndTime:   utils.BootOffsetToTime(e.EndTime),
-			Attributes: []attribute.KeyValue{
-				semconv.RPCSystemKey.String("grpc"),
-				semconv.RPCServiceKey.String(method),
-			},
-			ParentSpanContext: pscPtr,
-			SpanContext:       &sc,
-			TracerSchema:      semconv.SchemaURL,
+	event := &probe.SpanEvent{
+		SpanName:  method,
+		StartTime: utils.BootOffsetToTime(e.StartTime),
+		EndTime:   utils.BootOffsetToTime(e.EndTime),
+		Attributes: []attribute.KeyValue{
+			semconv.RPCSystemKey.String("grpc"),
+			semconv.RPCServiceKey.String(method),
+			semconv.RPCGRPCStatusCodeKey.Int(int(e.StatusCode)),
 		},
+		ParentSpanContext: pscPtr,
+		SpanContext:       &sc,
+		TracerSchema:      semconv.SchemaURL,
+	}
+
+	// Set server status codes per semconv:
+	// See https://github.com/open-telemetry/semantic-conventions/blob/02ecf0c71e9fa74d09d81c48e04a132db2b7060b/docs/rpc/grpc.md#grpc-status
+	if e.StatusCode == int32(codes.Unknown) ||
+		e.StatusCode == int32(codes.DeadlineExceeded) ||
+		e.StatusCode == int32(codes.Unimplemented) ||
+		e.StatusCode == int32(codes.Internal) ||
+		e.StatusCode == int32(codes.Unavailable) ||
+		e.StatusCode == int32(codes.DataLoss) {
+		event.Status = probe.Status{Code: otelcodes.Error}
 	}
+	return []*probe.SpanEvent{event}
 }

internal/test/e2e/grpc/main.go

@@ -14,8 +14,10 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	pb "google.golang.org/grpc/examples/helloworld/helloworld"
+	"google.golang.org/grpc/status"
 )
 
 const port = 1701
@@ -26,6 +28,9 @@ type server struct {
 
 func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
 	log.Printf("Received: %v", in.GetName())
+	if in.GetName() == "unimplemented" {
+		return nil, status.Error(codes.Unimplemented, "unimplmented")
+	}
 	return &pb.HelloReply{Message: "Hello " + in.GetName()}, nil
 }
 
@@ -68,6 +73,13 @@ func main() {
 	}
 	log.Printf("Greeting: %s", r.GetMessage())
 
+	// Contact the server expecting a server error
+	_, err = c.SayHello(ctx, &pb.HelloRequest{Name: "unimplemented"})
+	if err == nil {
+		log.Fatalf("expected an error but none was received")
+	}
+	log.Printf("received expected error: %+v", err)
+
 	s.GracefulStop()
 	<-done