Add request-timeout flag to control timeout for unary RPCs

Signed-off-by: Chance Zibolski <chance.zibolski@gmail.com>

cmd/common/config/flags.go

@@ -26,6 +26,7 @@ const (
 	KeyBasicAuthUsername = "basic-auth-username"  // string
 	KeyBasicAuthPassword = "basic-auth-password"  // string
 	KeyTimeout           = "timeout"              // time.Duration
+	KeyRequestTimeout    = "request-timeout"      // time.Duration
 )
 
 // GlobalFlags are flags that apply to any command.
@@ -47,6 +48,7 @@ func initGlobalFlags() {
 func initServerFlags() {
 	ServerFlags.String(KeyServer, defaults.ServerAddress, "Address of a Hubble server. Ignored when --input-file is provided.")
 	ServerFlags.Duration(KeyTimeout, defaults.DialTimeout, "Hubble server dialing timeout")
+	ServerFlags.Duration(KeyRequestTimeout, defaults.RequestTimeout, "Unary Request timeout. Only applies to non-streaming RPCs (ServerStatus, ListNodes, ListNamespaces).")
 	ServerFlags.Bool(
 		KeyTLS,
 		false,

cmd/common/conn/conn.go

@@ -9,7 +9,9 @@ import (
 	"strings"
 	"time"
 
+	"github.com/cilium/hubble/cmd/common/config"
 	"github.com/cilium/hubble/pkg/defaults"
+	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/timeout"
 	"github.com/spf13/viper"
 	"google.golang.org/grpc"
 )
@@ -26,6 +28,7 @@ func init() {
 		grpcOptionBlock,
 		grpcOptionFailOnNonTempDialError,
 		grpcOptionConnError,
+		grpcInterceptors,
 	)
 }
 
@@ -41,6 +44,10 @@ func grpcOptionConnError(_ *viper.Viper) (grpc.DialOption, error) {
 	return grpc.WithReturnConnectionError(), nil
 }
 
+func grpcInterceptors(vp *viper.Viper) (grpc.DialOption, error) {
+	return grpc.WithUnaryInterceptor(timeout.UnaryClientInterceptor(vp.GetDuration(config.KeyRequestTimeout))), nil
+}
+
 var grpcDialOptions []grpc.DialOption
 
 // Init initializes common connection options. It MUST be called prior to any

cmd/list/namespaces.go

@@ -24,8 +24,7 @@ func newNamespacesCommand(vp *viper.Viper) *cobra.Command {
 		Use:   "namespaces",
 		Short: "List namespaces with recent flows",
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
+			ctx := cmd.Context()
 			hubbleConn, err := conn.New(ctx, vp.GetString(config.KeyServer), vp.GetDuration(config.KeyTimeout))
 			if err != nil {
 				return err

cmd/list/node.go

@@ -31,13 +31,13 @@ func newNodeCommand(vp *viper.Viper) *cobra.Command {
 		Aliases: []string{"node"},
 		Short:   "List Hubble nodes",
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
+			ctx := cmd.Context()
 			hubbleConn, err := conn.New(ctx, vp.GetString(config.KeyServer), vp.GetDuration(config.KeyTimeout))
 			if err != nil {
 				return err
 			}
 			defer hubbleConn.Close()
+
 			return runListNodes(ctx, cmd, hubbleConn)
 		},
 	}

cmd/observe_help.txt

@@ -133,6 +133,7 @@ Flow Format Flags:
 Server Flags:
       --basic-auth-password string    Specify a password for basic auth
       --basic-auth-username string    Specify a username for basic auth
+      --request-timeout duration      Unary Request timeout. Only applies to non-streaming RPCs (ServerStatus, ListNodes, ListNamespaces). (default 12s)
       --server string                 Address of a Hubble server. Ignored when --input-file is provided. (default "localhost:4245")
       --timeout duration              Hubble server dialing timeout (default 5s)
       --tls                           Specify that TLS must be used when establishing a connection to a Hubble server.

cmd/status/status.go

@@ -14,7 +14,6 @@ import (
 	"github.com/cilium/hubble/cmd/common/config"
 	"github.com/cilium/hubble/cmd/common/conn"
 	"github.com/cilium/hubble/cmd/common/template"
-	"github.com/cilium/hubble/pkg/defaults"
 	"github.com/cilium/hubble/pkg/printer"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
@@ -35,14 +34,14 @@ func New(vp *viper.Viper) *cobra.Command {
 		Long: `Display shows the status of the Hubble server. This is intended as a basic
 connectivity health check.`,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
+			ctx := cmd.Context()
 			hubbleConn, err := conn.New(ctx, vp.GetString(config.KeyServer), vp.GetDuration(config.KeyTimeout))
 			if err != nil {
 				return err
 			}
 			defer hubbleConn.Close()
-			return runStatus(cmd.OutOrStdout(), hubbleConn)
+
+			return runStatus(ctx, cmd.OutOrStdout(), hubbleConn)
 		},
 	}
 
@@ -74,9 +73,9 @@ connectivity health check.`,
 	return statusCmd
 }
 
-func runStatus(out io.Writer, conn *grpc.ClientConn) error {
+func runStatus(ctx context.Context, out io.Writer, conn *grpc.ClientConn) error {
 	// get the standard GRPC health check to see if the server is up
-	healthy, status, err := getHC(conn)
+	healthy, status, err := getHC(ctx, conn)
 	if err != nil {
 		return fmt.Errorf("failed getting status: %v", err)
 	}
@@ -88,7 +87,7 @@ func runStatus(out io.Writer, conn *grpc.ClientConn) error {
 	}
 
 	// if the server is up, lets try to get hubble specific status
-	ss, err := getStatus(conn)
+	ss, err := getStatus(ctx, conn)
 	if err != nil {
 		return fmt.Errorf("failed to get hubble server status: %v", err)
 	}
@@ -115,10 +114,7 @@ func runStatus(out io.Writer, conn *grpc.ClientConn) error {
 	return p.Close()
 }
 
-func getHC(conn *grpc.ClientConn) (healthy bool, status string, err error) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaults.RequestTimeout)
-	defer cancel()
-
+func getHC(ctx context.Context, conn *grpc.ClientConn) (healthy bool, status string, err error) {
 	req := &healthpb.HealthCheckRequest{Service: v1.ObserverServiceName}
 	resp, err := healthpb.NewHealthClient(conn).Check(ctx, req)
 	if err != nil {
@@ -130,10 +126,7 @@ func getHC(conn *grpc.ClientConn) (healthy bool, status string, err error) {
 	return true, "Ok", nil
 }
 
-func getStatus(conn *grpc.ClientConn) (*observerpb.ServerStatusResponse, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaults.RequestTimeout)
-	defer cancel()
-
+func getStatus(ctx context.Context, conn *grpc.ClientConn) (*observerpb.ServerStatusResponse, error) {
 	req := &observerpb.ServerStatusRequest{}
 	res, err := observerpb.NewObserverClient(conn).ServerStatus(ctx, req)
 	if err != nil {

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/cilium/cilium v1.15.0-pre.1.0.20231016085253-84f5b169c565
 	github.com/fatih/color v1.15.0
 	github.com/google/go-cmp v0.6.0
+	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1
 	github.com/spf13/cast v1.5.1
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace