mirror of
https://github.com/gravitational/teleport.git
synced 2026-09-24 16:17:11 +08:00
Add gRPC error interceptors to API client (#30578)
* Move gRPC error intercetpors to api/utils/grpc/interceptors. * Use error interceptors in api client and mock server. * Apply suggestions from CR. * Unwrap FromGRPC errors in middleware. * Use gRPC auth service in tests instead of external example service. * It's gRPC!!! * Fix unit test. * Add error interceptor to proxy client. * Fix merge conflict.
This commit is contained in:
@@ -38,6 +38,7 @@ import (
|
||||
transportv1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/transport/v1"
|
||||
"github.com/gravitational/teleport/api/metadata"
|
||||
"github.com/gravitational/teleport/api/observability/tracing"
|
||||
"github.com/gravitational/teleport/api/utils/grpc/interceptors"
|
||||
)
|
||||
|
||||
// ClientConfig contains configuration needed for a Client
|
||||
@@ -274,12 +275,14 @@ func newGRPCClient(ctx context.Context, cfg *ClientConfig) (_ *Client, err error
|
||||
append(cfg.UnaryInterceptors,
|
||||
otelgrpc.UnaryClientInterceptor(),
|
||||
metadata.UnaryClientInterceptor,
|
||||
interceptors.GRPCClientUnaryErrorInterceptor,
|
||||
)...,
|
||||
),
|
||||
grpc.WithChainStreamInterceptor(
|
||||
append(cfg.StreamInterceptors,
|
||||
otelgrpc.StreamClientInterceptor(),
|
||||
metadata.StreamClientInterceptor,
|
||||
interceptors.GRPCClientStreamErrorInterceptor,
|
||||
)...,
|
||||
),
|
||||
}, cfg.DialOpts...)...,
|
||||
|
||||
@@ -20,7 +20,6 @@ import (
|
||||
"sync"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/gravitational/trace/trail"
|
||||
"golang.org/x/crypto/ssh/agent"
|
||||
"google.golang.org/grpc/peer"
|
||||
|
||||
@@ -51,7 +50,7 @@ func NewClient(client transportv1pb.TransportServiceClient) (*Client, error) {
|
||||
func (c *Client) ClusterDetails(ctx context.Context) (*transportv1pb.ClusterDetails, error) {
|
||||
resp, err := c.clt.GetClusterDetails(ctx, &transportv1pb.GetClusterDetailsRequest{})
|
||||
if err != nil {
|
||||
return nil, trail.FromGRPC(err)
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
|
||||
return resp.Details, nil
|
||||
@@ -62,11 +61,11 @@ func (c *Client) ClusterDetails(ctx context.Context) (*transportv1pb.ClusterDeta
|
||||
func (c *Client) DialCluster(ctx context.Context, cluster string, src net.Addr) (net.Conn, error) {
|
||||
stream, err := c.clt.ProxyCluster(ctx)
|
||||
if err != nil {
|
||||
return nil, trail.FromGRPC(err, "unable to establish proxy stream")
|
||||
return nil, trace.Wrap(err, "unable to establish proxy stream")
|
||||
}
|
||||
|
||||
if err := stream.Send(&transportv1pb.ProxyClusterRequest{Cluster: cluster}); err != nil {
|
||||
return nil, trail.FromGRPC(err, "failed to send cluster request")
|
||||
return nil, trace.Wrap(err, "failed to send cluster request")
|
||||
}
|
||||
|
||||
streamRW, err := streamutils.NewReadWriter(clusterStream{stream: stream})
|
||||
@@ -115,19 +114,19 @@ func (c clusterStream) Close() error {
|
||||
func (c *Client) DialHost(ctx context.Context, hostport, cluster string, src net.Addr, keyring agent.ExtendedAgent) (net.Conn, *transportv1pb.ClusterDetails, error) {
|
||||
stream, err := c.clt.ProxySSH(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, trail.FromGRPC(err, "unable to establish proxy stream")
|
||||
return nil, nil, trace.Wrap(err, "unable to establish proxy stream")
|
||||
}
|
||||
|
||||
if err := stream.Send(&transportv1pb.ProxySSHRequest{DialTarget: &transportv1pb.TargetHost{
|
||||
HostPort: hostport,
|
||||
Cluster: cluster,
|
||||
}}); err != nil {
|
||||
return nil, nil, trail.FromGRPC(err, "failed to send dial target request")
|
||||
return nil, nil, trace.Wrap(err, "failed to send dial target request")
|
||||
}
|
||||
|
||||
resp, err := stream.Recv()
|
||||
if err != nil {
|
||||
return nil, nil, trail.FromGRPC(err, "failed to receive cluster details response")
|
||||
return nil, nil, trace.Wrap(err, "failed to receive cluster details response")
|
||||
}
|
||||
|
||||
// create streams for ssh and agent protocol
|
||||
@@ -161,8 +160,8 @@ func (c *Client) DialHost(ctx context.Context, hostport, cluster string, src net
|
||||
for {
|
||||
req, err := stream.Recv()
|
||||
if err != nil {
|
||||
sshStream.errorC <- trail.FromGRPC(err)
|
||||
agentStream.errorC <- trail.FromGRPC(err)
|
||||
sshStream.errorC <- trace.Wrap(err)
|
||||
agentStream.errorC <- trace.Wrap(err)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -37,6 +37,7 @@ import (
|
||||
"google.golang.org/grpc/test/bufconn"
|
||||
|
||||
transportv1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/transport/v1"
|
||||
"github.com/gravitational/teleport/api/utils/grpc/interceptors"
|
||||
streamutils "github.com/gravitational/teleport/api/utils/grpc/stream"
|
||||
)
|
||||
|
||||
@@ -537,6 +538,8 @@ func newServer(t *testing.T, srv transportv1pb.TransportServiceServer) testPack
|
||||
}),
|
||||
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(1000)),
|
||||
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
grpc.WithUnaryInterceptor(interceptors.GRPCClientUnaryErrorInterceptor),
|
||||
grpc.WithStreamInterceptor(interceptors.GRPCClientStreamErrorInterceptor),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
|
||||
Reference in New Issue
Block a user