fix(grpc): honor RPC status, timeout settings and reflection headers

internal/igrpc/disabled_timeout_test.go

@@ -0,0 +1,31 @@
+package igrpc_test
+
+import (
+	"context"
+	"github.com/kavix/kurl/internal/igrpc"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/reflection"
+	"net"
+	"testing"
+	"time"
+)
+
+func TestRunAllowsDisabledTimeout(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := grpc.NewServer()
+	grpc_health_v1.RegisterHealthServer(server, health.NewServer())
+	reflection.Register(server)
+	go func() { _ = server.Serve(listener) }()
+	defer server.Stop()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	err = igrpc.Run(ctx, igrpc.Options{URL: "grpc://" + listener.Addr().String(), Method: "grpc.health.v1.Health/Check", Data: `{}`, Timeout: 0})
+	if err != nil {
+		t.Fatalf("disabled timeout prevented RPC: %v", err)
+	}
+}

internal/igrpc/grpc.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
 )
 
 type Options struct {
@@ -55,8 +56,12 @@ func Run(ctx context.Context, opts Options) error {
 		creds = insecure.NewCredentials()
 	}
 
-	dialCtx, cancel := context.WithTimeout(ctx, opts.Timeout)
-	defer cancel()
+	dialCtx := ctx
+	if opts.Timeout > 0 {
+		var cancel context.CancelFunc
+		dialCtx, cancel = context.WithTimeout(ctx, opts.Timeout)
+		defer cancel()
+	}
 
 	cc, err := grpc.DialContext(dialCtx, target, grpc.WithTransportCredentials(creds))
 	if err != nil {
@@ -72,9 +77,10 @@ func Run(ctx context.Context, opts Options) error {
 			return fmt.Errorf("failed to process proto file: %w", err)
 		}
 	} else {
-		refClient := grpcreflect.NewClientAuto(dialCtx, cc)
+		reflectionCtx := metadata.NewOutgoingContext(dialCtx, grpcurl.MetadataFromHeaders(opts.Headers))
+		refClient := grpcreflect.NewClientAuto(reflectionCtx, cc)
 		defer refClient.Reset()
-		descSource = grpcurl.DescriptorSourceFromServer(dialCtx, refClient)
+		descSource = grpcurl.DescriptorSourceFromServer(reflectionCtx, refClient)
 	}
 
 	if opts.ListServices {
@@ -118,13 +124,14 @@ func Run(ctx context.Context, opts Options) error {
 		h.VerbosityLevel = 1
 	}
 
-	err = grpcurl.InvokeRPC(ctx, descSource, cc, opts.Method, opts.Headers, h, rf.Next)
+	err = grpcurl.InvokeRPC(dialCtx, descSource, cc, opts.Method, opts.Headers, h, rf.Next)
 	if err != nil {
 		return fmt.Errorf("rpc error: %w", err)
 	}
 
 	if h.Status.Code() != 0 {
 		grpcurl.PrintStatus(os.Stderr, h.Status, formatter)
+		return h.Status.Err()
 	}
 
 	return nil

internal/igrpc/reflection_headers_test.go

@@ -0,0 +1,56 @@
+package igrpc_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/kavix/kurl/internal/igrpc"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/reflection"
+	"google.golang.org/grpc/status"
+)
+
+func TestRunSendsHeadersToReflection(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	authenticated := func(ctx context.Context) bool {
+		values := metadata.ValueFromIncomingContext(ctx, "authorization")
+		return len(values) == 1 && values[0] == "Bearer test-token"
+	}
+	server := grpc.NewServer(
+		grpc.StreamInterceptor(func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+			if !authenticated(stream.Context()) {
+				return status.Error(codes.Unauthenticated, "reflection requires authorization")
+			}
+			return handler(srv, stream)
+		}),
+		grpc.UnaryInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+			if !authenticated(ctx) {
+				return nil, status.Error(codes.Unauthenticated, "RPC requires authorization")
+			}
+			return handler(ctx, req)
+		}),
+	)
+	grpc_health_v1.RegisterHealthServer(server, health.NewServer())
+	reflection.Register(server)
+	go func() { _ = server.Serve(listener) }()
+	defer server.Stop()
+	for _, list := range []bool{true, false} {
+		err := igrpc.Run(context.Background(), igrpc.Options{
+			URL: "grpc://" + listener.Addr().String(), ListServices: list,
+			Method: "grpc.health.v1.Health/Check", Data: `{}`,
+			Headers: []string{"Authorization: Bearer test-token"}, Timeout: 5 * time.Second,
+		})
+		if err != nil {
+			t.Errorf("list=%v: %v", list, err)
+		}
+	}
+}

internal/igrpc/status_test.go

@@ -0,0 +1,35 @@
+package igrpc_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/kavix/kurl/internal/igrpc"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/reflection"
+	"google.golang.org/grpc/status"
+)
+
+func TestRunReturnsRPCStatus(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := grpc.NewServer()
+	grpc_health_v1.RegisterHealthServer(server, health.NewServer())
+	reflection.Register(server)
+	go func() { _ = server.Serve(listener) }()
+	defer server.Stop()
+	err = igrpc.Run(context.Background(), igrpc.Options{
+		URL: "grpc://" + listener.Addr().String(), Method: "grpc.health.v1.Health/Check",
+		Data: `{"service":"missing"}`, Timeout: 5 * time.Second,
+	})
+	if status.Code(err) != codes.NotFound {
+		t.Fatalf("got %v (%v), want NotFound", err, status.Code(err))
+	}
+}

internal/igrpc/timeout_test.go

@@ -0,0 +1,61 @@
+package igrpc_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/kavix/kurl/internal/igrpc"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/reflection"
+	"google.golang.org/grpc/status"
+)
+
+type waitingHealthServer struct {
+	grpc_health_v1.UnimplementedHealthServer
+	remaining chan time.Duration
+}
+
+func (s *waitingHealthServer) Check(ctx context.Context, _ *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		s.remaining <- time.Hour
+	} else {
+		s.remaining <- time.Until(deadline)
+	}
+	<-ctx.Done()
+	return nil, status.FromContextError(ctx.Err()).Err()
+}
+
+func TestRunAppliesTimeoutToInvocation(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := grpc.NewServer()
+	healthServer := &waitingHealthServer{remaining: make(chan time.Duration, 1)}
+	grpc_health_v1.RegisterHealthServer(server, healthServer)
+	reflection.Register(server)
+	go func() { _ = server.Serve(listener) }()
+	defer server.Stop()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	err = igrpc.Run(ctx, igrpc.Options{
+		URL: "grpc://" + listener.Addr().String(), Method: "grpc.health.v1.Health/Check",
+		Data: `{}`, Timeout: time.Second,
+	})
+	if status.Code(err) != codes.DeadlineExceeded {
+		t.Fatalf("got %v, want deadline exceeded", err)
+	}
+	select {
+	case remaining := <-healthServer.remaining:
+		if remaining > time.Second {
+			t.Errorf("RPC received a %s deadline, exceeding the configured one-second timeout", remaining)
+		}
+	default:
+		t.Fatal("RPC did not reach the server")
+	}
+}