galaxy-game/gateway/internal/grpcapi/observability.go

package grpcapi

import (
	"context"
	"errors"
	"path"
	"time"

	"galaxy/gateway/internal/logging"
	"galaxy/gateway/internal/telemetry"
	edgev1 "galaxy/gateway/proto/edge/v1"

	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// recordEdgeRequest emits the structured log entry and the
// `gateway.authenticated_grpc.*` metric pair for one authenticated edge
// request or stream outcome. The transport parameter labels the wire
// protocol the request travelled over (`connect`, `grpc`, or `grpc-web`),
// preserving stable observability semantics across the unified Connect-go
// listener.
func recordEdgeRequest(logger *zap.Logger, metrics *telemetry.Runtime, ctx context.Context, transport string, fullMethod string, req any, resp any, err error, duration time.Duration, streamKind string) {
	rpcMethod := path.Base(fullMethod)
	messageType, requestID, traceID := envelopeFieldsFromRequest(req)
	resultCode := resultCodeFromResponse(resp)
	grpcCode, grpcMessage, outcome := outcomeFromError(err)
	rejectReason := telemetry.RejectReason(outcome)

	attrs := []attribute.KeyValue{
		attribute.String("rpc_method", rpcMethod),
		attribute.String("message_type", messageType),
		attribute.String("edge_outcome", string(outcome)),
	}
	if resultCode != "" {
		attrs = append(attrs, attribute.String("result_code", resultCode))
	}
	if rejectReason != "" {
		attrs = append(attrs, attribute.String("reject_reason", rejectReason))
	}
	metrics.RecordAuthenticatedGRPC(ctx, attrs, duration)

	fields := []zap.Field{
		zap.String("component", "authenticated_grpc"),
		zap.String("transport", transport),
		zap.String("stream_kind", streamKind),
		zap.String("rpc_method", rpcMethod),
		zap.String("message_type", messageType),
		zap.String("grpc_code", grpcCode.String()),
		zap.Float64("duration_ms", float64(duration.Microseconds())/1000),
		zap.String("request_id", requestID),
		zap.String("trace_id", traceID),
		zap.String("peer_ip", peerIPFromContext(ctx)),
		zap.String("edge_outcome", string(outcome)),
	}
	if resultCode != "" {
		fields = append(fields, zap.String("result_code", resultCode))
	}
	if rejectReason != "" {
		fields = append(fields, zap.String("reject_reason", rejectReason))
	}
	if grpcMessage != "" {
		fields = append(fields, zap.String("grpc_message", grpcMessage))
	}
	fields = append(fields, logging.TraceFieldsFromContext(ctx)...)

	switch outcome {
	case telemetry.EdgeOutcomeSuccess:
		logger.Info("authenticated edge request completed", fields...)
	case telemetry.EdgeOutcomeBackendUnavailable, telemetry.EdgeOutcomeDownstreamUnavailable, telemetry.EdgeOutcomeInternalError:
		logger.Error("authenticated edge request failed", fields...)
	default:
		logger.Warn("authenticated edge request rejected", fields...)
	}
}

func envelopeFieldsFromRequest(req any) (messageType string, requestID string, traceID string) {
	switch typed := req.(type) {
	case *edgev1.ExecuteCommandRequest:
		return typed.GetMessageType(), typed.GetRequestId(), typed.GetTraceId()
	case *edgev1.SubscribeEventsRequest:
		return typed.GetMessageType(), typed.GetRequestId(), typed.GetTraceId()
	default:
		return "", "", ""
	}
}

func resultCodeFromResponse(resp any) string {
	typed, ok := resp.(*edgev1.ExecuteCommandResponse)
	if !ok {
		return ""
	}

	return typed.GetResultCode()
}

func outcomeFromError(err error) (codes.Code, string, telemetry.EdgeOutcome) {
	switch {
	case err == nil:
		return codes.OK, "", telemetry.EdgeOutcomeSuccess
	case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
		return codes.Canceled, err.Error(), telemetry.EdgeOutcomeSuccess
	default:
		grpcStatus := status.Convert(err)
		return grpcStatus.Code(), grpcStatus.Message(), telemetry.OutcomeFromGRPCStatus(grpcStatus.Code(), grpcStatus.Message())
	}
}