Adding retries for gRPC calls (#1248)

Author: akirillov

Gopkg.lock

@@ -115,3 +115,7 @@ required = [
 [[prune.project]]
   name = "k8s.io/gengo"
   unused-packages = false
+
+[[constraint]]
+  name = "github.com/grpc-ecosystem/go-grpc-middleware"
+  version = "1.2.0"

Gopkg.toml

@@ -1,6 +1,9 @@
 package consts
 
-import "github.com/kubeflow/katib/pkg/util/v1alpha3/env"
+import (
+	"github.com/kubeflow/katib/pkg/util/v1alpha3/env"
+	"time"
+)
 
 const (
 	// ConfigExperimentSuggestionName is the config name of the
@@ -35,6 +38,11 @@ const (
 	// which is used to run healthz check using grpc probe.
 	DefaultGRPCService = "manager.v1alpha3.Suggestion"
 
+	// DefaultGRPCRetryAttempts is the the maximum number of retries for gRPC calls
+	DefaultGRPCRetryAttempts = 10
+	// DefaultGRPCRetryPeriod is a fixed period of time between gRPC call retries
+	DefaultGRPCRetryPeriod = 3 * time.Second
+
 	// DefaultKatibNamespaceEnvName is the default env name of katib namespace
 	DefaultKatibNamespaceEnvName = "KATIB_CORE_NAMESPACE"
 	// DefaultKatibComposerEnvName is the default env name of katib suggestion composer

pkg/controller.v1alpha3/consts/const.go

@@ -5,6 +5,9 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/kubeflow/katib/pkg/controller.v1alpha3/consts"
+
+	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -105,7 +108,15 @@ func (g *General) SyncAssignments(
 func (g *General) ValidateAlgorithmSettings(instance *suggestionsv1alpha3.Suggestion, e *experimentsv1alpha3.Experiment) error {
 	logger := log.WithValues("Suggestion", types.NamespacedName{Name: instance.GetName(), Namespace: instance.GetNamespace()})
 	endpoint := util.GetAlgorithmEndpoint(instance)
-	conn, err := grpc.Dial(endpoint, grpc.WithInsecure())
+
+	callOpts := []grpc_retry.CallOption{
+		grpc_retry.WithBackoff(grpc_retry.BackoffLinear(consts.DefaultGRPCRetryPeriod)),
+		grpc_retry.WithMax(consts.DefaultGRPCRetryAttempts),
+	}
+	conn, err := grpc.Dial(endpoint, grpc.WithInsecure(),
+		grpc.WithStreamInterceptor(grpc_retry.StreamClientInterceptor(callOpts...)),
+		grpc.WithUnaryInterceptor(grpc_retry.UnaryClientInterceptor(callOpts...)),
+	)
 	if err != nil {
 		return err
 	}
@@ -118,6 +129,7 @@ func (g *General) ValidateAlgorithmSettings(instance *suggestionsv1alpha3.Sugges
 	request := &suggestionapi.ValidateAlgorithmSettingsRequest{
 		Experiment: g.ConvertExperiment(e),
 	}
+
 	// See https://github.com/grpc/grpc-go/issues/2636
 	// See https://github.com/grpc/grpc-go/pull/2503
 	_, err = client.ValidateAlgorithmSettings(ctx, request, grpc.WaitForReady(true))

pkg/controller.v1alpha3/suggestion/suggestionclient/suggestionclient.go

@@ -1,6 +1,9 @@
 package consts
 
-import "github.com/kubeflow/katib/pkg/util/v1beta1/env"
+import (
+	"github.com/kubeflow/katib/pkg/util/v1beta1/env"
+	"time"
+)
 
 const (
 	// ConfigExperimentSuggestionName is the config name of the
@@ -35,6 +38,11 @@ const (
 	// which is used to run healthz check using grpc probe.
 	DefaultGRPCService = "manager.v1beta1.Suggestion"
 
+	// DefaultGRPCRetryAttempts is the the maximum number of retries for gRPC calls
+	DefaultGRPCRetryAttempts = 10
+	// DefaultGRPCRetryPeriod is a fixed period of time between gRPC call retries
+	DefaultGRPCRetryPeriod = 3 * time.Second
+
 	// DefaultKatibNamespaceEnvName is the default env name of katib namespace
 	DefaultKatibNamespaceEnvName = "KATIB_CORE_NAMESPACE"
 	// DefaultKatibComposerEnvName is the default env name of katib suggestion composer

pkg/controller.v1beta1/consts/const.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"time"
 
+	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
+
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -110,7 +112,15 @@ func (g *General) SyncAssignments(
 func (g *General) ValidateAlgorithmSettings(instance *suggestionsv1beta1.Suggestion, e *experimentsv1beta1.Experiment) error {
 	logger := log.WithValues("Suggestion", types.NamespacedName{Name: instance.GetName(), Namespace: instance.GetNamespace()})
 	endpoint := util.GetAlgorithmEndpoint(instance)
-	conn, err := grpc.Dial(endpoint, grpc.WithInsecure())
+
+	callOpts := []grpc_retry.CallOption{
+		grpc_retry.WithBackoff(grpc_retry.BackoffLinear(consts.DefaultGRPCRetryPeriod)),
+		grpc_retry.WithMax(consts.DefaultGRPCRetryAttempts),
+	}
+	conn, err := grpc.Dial(endpoint, grpc.WithInsecure(),
+		grpc.WithStreamInterceptor(grpc_retry.StreamClientInterceptor(callOpts...)),
+		grpc.WithUnaryInterceptor(grpc_retry.UnaryClientInterceptor(callOpts...)),
+	)
 	if err != nil {
 		return err
 	}