Ilia Denisov
118f7c17a2
Replace the native-gRPC server bootstrap with a single `connectrpc.com/connect` HTTP/h2c listener. Connect-Go natively serves Connect, gRPC, and gRPC-Web on the same port, so browsers can now reach the authenticated surface without giving up the gRPC framing native and desktop clients may use later. The decorator stack (envelope → session → payload-hash → signature → freshness/replay → rate-limit → routing/push) is reused unchanged behind a small Connect → gRPC adapter and a `grpc.ServerStream` shim around `*connect.ServerStream`. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
110 lines
3.8 KiB
Go
110 lines
3.8 KiB
Go
package grpcapi
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"path"
|
|
"time"
|
|
|
|
"galaxy/gateway/internal/logging"
|
|
"galaxy/gateway/internal/telemetry"
|
|
gatewayv1 "galaxy/gateway/proto/galaxy/gateway/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 *gatewayv1.ExecuteCommandRequest:
|
|
return typed.GetMessageType(), typed.GetRequestId(), typed.GetTraceId()
|
|
case *gatewayv1.SubscribeEventsRequest:
|
|
return typed.GetMessageType(), typed.GetRequestId(), typed.GetTraceId()
|
|
default:
|
|
return "", "", ""
|
|
}
|
|
}
|
|
|
|
func resultCodeFromResponse(resp any) string {
|
|
typed, ok := resp.(*gatewayv1.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())
|
|
}
|
|
}
|