From e548acc71208820c5c465b684ce1f4f2c15a87cf Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Tue, 8 Apr 2025 17:03:52 +0530 Subject: [PATCH 1/8] generic retry for streams --- internal/service/component.go | 74 ++-------- internal/service/service.go | 215 ++++------------------------- internal/service/stream_handler.go | 90 ++++++++++++ pkg/util/util.go | 12 ++ 4 files changed, 142 insertions(+), 249 deletions(-) create mode 100644 internal/service/stream_handler.go diff --git a/internal/service/component.go b/internal/service/component.go index 95c2f327..e50d50db 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -2,8 +2,6 @@ package service import ( "context" - "fmt" - "io" "time" "github.com/briandowns/spinner" @@ -36,69 +34,17 @@ func (e *Component) OperateComponent(ctx *context.Context, request *serviceProto return err } - var message string - var retries = 0 - - responseChan := make(chan *serviceProto.OperateServiceResponse) - errorChan := make(chan error) - go func(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - default: - response, err := stream.Recv() - if err != nil { - errorChan <- err - } - responseChan <- response - } - } - }(*requestCtx) - - for { - recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout) - - select { - case <-recvCtx.Done(): - spinnerInstance.Stop() - cancel() - if !util.CanPerformRetry(retries, constant.MaxRetries) { - return nil - } - stream, err = reconnectOperateStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case err := <-errorChan: - spinnerInstance.Stop() - cancel() - if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { - if err != io.EOF { - return err - } else if err == io.EOF { - log.Info(message) - } - return nil - } - stream, err = reconnectOperateStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case response := <-responseChan: - spinnerInstance.Stop() - cancel() - if response != nil { - message = util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()}) - logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()}) - spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) - spinnerInstance.Start() - retries = 0 - } - } + reconnect := func() (serviceProto.ServiceService_OperateServiceClient, error) { + return reconnectOperateStream(client, requestCtx, request, stream) } + + generateResponse := func(response *serviceProto.OperateServiceResponse) string { + message := util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()}) + logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()}) + return message + } + + return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.OperateServiceRequest, stream serviceProto.ServiceService_OperateServiceClient) (serviceProto.ServiceService_OperateServiceClient, error) { diff --git a/internal/service/service.go b/internal/service/service.go index 26de94b7..a0fa05de 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -40,69 +40,17 @@ func (e *Service) DeployService(ctx *context.Context, request *serviceProto.Depl return err } - var message string - var retries = 0 - - responseChan := make(chan *serviceProto.DeployServiceResponse) - errorChan := make(chan error) - go func(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - default: - response, err := stream.Recv() - if err != nil { - errorChan <- err - } - responseChan <- response - } - } - }(*requestCtx) - - for { - recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout) + reconnect := func() (serviceProto.ServiceService_DeployServiceClient, error) { + return reconnectDeployServiceStream(client, requestCtx, request, stream) + } - select { - case <-recvCtx.Done(): - spinnerInstance.Stop() - cancel() - if !util.CanPerformRetry(retries, constant.MaxRetries) { - return nil - } - stream, err = reconnectDeployServiceStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case err := <-errorChan: - spinnerInstance.Stop() - cancel() - if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { - if err != io.EOF { - return err - } else if err == io.EOF { - log.Info(message) - } - return nil - } - stream, err = reconnectDeployServiceStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case response := <-responseChan: - spinnerInstance.Stop() - cancel() - if response != nil { - message = util.GenerateResponseMessage(response.GetServiceResponse()) - logFailedComponentMessagesOnce(response.GetServiceResponse()) - spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) - spinnerInstance.Start() - retries = 0 // Reset retries on successful response - } - } + generateResponse := func(response *serviceProto.DeployServiceResponse) string { + message := util.GenerateResponseMessage(response.GetServiceResponse()) + logFailedComponentMessagesOnce(response.GetServiceResponse()) + return message } + + return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } func logFailedComponentMessagesOnce(response *serviceProto.ServiceResponse) { @@ -214,74 +162,22 @@ func (e *Service) DeployReleasedService(ctx *context.Context, request *servicePr if err != nil { return err } + reconnect := func() (serviceProto.ServiceService_DeployReleasedServiceClient, error) { + return reconnectDeployReleasedServiceStream(client, requestCtx, request, stream) + } - var message string - var retries = 0 - - responseChan := make(chan *serviceProto.DeployReleasedServiceResponse) - errorChan := make(chan error) - go func(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - default: - response, err := stream.Recv() - if err != nil { - errorChan <- err - } - responseChan <- response - } - } - }(*requestCtx) - for { - recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout) - - select { - case <-recvCtx.Done(): - spinnerInstance.Stop() - cancel() - if !util.CanPerformRetry(retries, constant.MaxRetries) { - return nil - } - stream, err = reconnectDeployReleasedServiceStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case err := <-errorChan: - spinnerInstance.Stop() - cancel() - if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { - if err != io.EOF { - return err - } else if err == io.EOF { - log.Info(message) - } - return nil - } - stream, err = reconnectDeployReleasedServiceStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case response := <-responseChan: - spinnerInstance.Stop() - cancel() - if response != nil { - message = response.ServiceResponse.Message - message += fmt.Sprintf("\n Service %s %s", response.ServiceResponse.ServiceStatus.ServiceAction, response.ServiceResponse.ServiceStatus) - for _, compMessage := range response.ServiceResponse.ComponentsStatus { - message += fmt.Sprintf("\n Component %s %s %s", compMessage.ComponentName, compMessage.ComponentAction, compMessage.ComponentStatus) - } - logFailedComponentMessagesOnce(response.GetServiceResponse()) - spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) - spinnerInstance.Start() - retries = 0 // Reset retries on successful response - } + generateResponse := func(response *serviceProto.DeployReleasedServiceResponse) string { + message := response.ServiceResponse.Message + message += fmt.Sprintf("\n Service %s %s", response.ServiceResponse.ServiceStatus.ServiceAction, response.ServiceResponse.ServiceStatus) + for _, compMessage := range response.ServiceResponse.ComponentsStatus { + message += fmt.Sprintf("\n Component %s %s %s", compMessage.ComponentName, compMessage.ComponentAction, compMessage.ComponentStatus) } + logFailedComponentMessagesOnce(response.GetServiceResponse()) + return message } + return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + } // UndeployService undeploy service @@ -349,68 +245,17 @@ func (e *Service) OperateService(ctx *context.Context, request *serviceProto.Ope return err } - var message string - var retries = 0 - - responseChan := make(chan *serviceProto.OperateServiceResponse) - errorChan := make(chan error) - go func(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - default: - response, err := stream.Recv() - if err != nil { - errorChan <- err - } - responseChan <- response - } - } - }(*requestCtx) - for { - recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout) + reconnect := func() (serviceProto.ServiceService_OperateServiceClient, error) { + return reconnectOperateStream(client, requestCtx, request, stream) + } - select { - case <-recvCtx.Done(): - spinnerInstance.Stop() - cancel() - if !util.CanPerformRetry(retries, constant.MaxRetries) { - return nil - } - stream, err = reconnectOperateStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case err := <-errorChan: - spinnerInstance.Stop() - cancel() - if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { - if err != io.EOF { - return err - } else if err == io.EOF { - log.Info(message) - } - return nil - } - stream, err = reconnectOperateStream(client, requestCtx, request, stream) - if err != nil { - return nil - } - retries++ - case response := <-responseChan: - spinnerInstance.Stop() - cancel() - if response != nil { - message = util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()}) - logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()}) - spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) - spinnerInstance.Start() - retries = 0 - } - } + generateResponse := func(response *serviceProto.OperateServiceResponse) string { + message := util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()}) + logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()}) + return message } + + return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } // ListService deploys service diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go new file mode 100644 index 00000000..331399d6 --- /dev/null +++ b/internal/service/stream_handler.go @@ -0,0 +1,90 @@ +package service + +import ( + "context" + "fmt" + "io" + + "github.com/briandowns/spinner" + "github.com/dream11/odin/pkg/constant" + "github.com/dream11/odin/pkg/util" + log "github.com/sirupsen/logrus" +) + +// StreamReceiverInterface defines the Recv method that the stream must implement +type StreamReceiverInterface[R any] interface { + Recv() (R, error) +} + +// ReconnectFunc defines a function type for reconnecting the stream +type ReconnectFunc[S any] func() (S, error) + +// GenerateResponse defines a function type for generating a response message +type GenerateResponse[R any] func(response R) string + +// handleStreamReponse handles the stream response, retries on errors, and updates the spinner +func handleStreamReponse[S StreamReceiverInterface[R], R any](stream S, requestCtx *context.Context, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { + var message string + var retries = 0 + errorChan := make(chan error) + responseChan := make(chan R) + + go func(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + default: + response, err := stream.Recv() + if err != nil { + errorChan <- err + } + responseChan <- response + } + } + }(*requestCtx) + + for { + recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout) + + select { + case <-recvCtx.Done(): + spinnerInstance.Stop() + cancel() + if !util.CanPerformRetry(retries, constant.MaxRetries) { + return nil + } + var er error + stream, er = reconnect() + if er != nil { + return nil + } + retries++ + case err := <-errorChan: + spinnerInstance.Stop() + cancel() + if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { + if err != io.EOF { + return err + } else if err == io.EOF { + log.Info(message) + } + return nil + } + stream, err = reconnect() + if err != nil { + return nil + } + retries++ + case response := <-responseChan: + spinnerInstance.Stop() + cancel() + if !util.IsNil(response) { + message = generateResponse(response) + spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) + spinnerInstance.Start() + retries = 0 + } + } + } +} diff --git a/pkg/util/util.go b/pkg/util/util.go index 59b8b85c..c6ba1c31 100644 --- a/pkg/util/util.go +++ b/pkg/util/util.go @@ -8,6 +8,7 @@ import ( "io" "net" "os" + "reflect" "strconv" "strings" "time" @@ -213,3 +214,14 @@ func CanPerformRetry( return true } + +// IsNil checks if a value is nil +func IsNil[T any](v T) bool { + val := reflect.ValueOf(v) + switch val.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return val.IsNil() + default: + return false + } +} From 768049dd8cc55119332120ba8dd5505ea6280c0e Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Wed, 9 Apr 2025 12:17:59 +0530 Subject: [PATCH 2/8] renamed function --- internal/service/component.go | 2 +- internal/service/service.go | 6 +++--- internal/service/stream_handler.go | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/internal/service/component.go b/internal/service/component.go index e50d50db..6bfc533c 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -44,7 +44,7 @@ func (e *Component) OperateComponent(ctx *context.Context, request *serviceProto return message } - return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.OperateServiceRequest, stream serviceProto.ServiceService_OperateServiceClient) (serviceProto.ServiceService_OperateServiceClient, error) { diff --git a/internal/service/service.go b/internal/service/service.go index a0fa05de..bd78613f 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -50,7 +50,7 @@ func (e *Service) DeployService(ctx *context.Context, request *serviceProto.Depl return message } - return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } func logFailedComponentMessagesOnce(response *serviceProto.ServiceResponse) { @@ -176,7 +176,7 @@ func (e *Service) DeployReleasedService(ctx *context.Context, request *servicePr return message } - return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } @@ -255,7 +255,7 @@ func (e *Service) OperateService(ctx *context.Context, request *serviceProto.Ope return message } - return handleStreamReponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) } // ListService deploys service diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go index 331399d6..d46115c4 100644 --- a/internal/service/stream_handler.go +++ b/internal/service/stream_handler.go @@ -22,8 +22,8 @@ type ReconnectFunc[S any] func() (S, error) // GenerateResponse defines a function type for generating a response message type GenerateResponse[R any] func(response R) string -// handleStreamReponse handles the stream response, retries on errors, and updates the spinner -func handleStreamReponse[S StreamReceiverInterface[R], R any](stream S, requestCtx *context.Context, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { +// handleStreamResponse handles the stream response, retries on errors, and updates the spinner +func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, requestCtx *context.Context, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { var message string var retries = 0 errorChan := make(chan error) From 01fa702349a382adc93a87ac65eeaed387d6ece8 Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Wed, 9 Apr 2025 19:26:34 +0530 Subject: [PATCH 3/8] removed go routine and select block --- internal/service/client.go | 8 ++++ internal/service/component.go | 14 ++++--- internal/service/service.go | 43 ++++++++++++-------- internal/service/stream_handler.go | 63 ++++++------------------------ pkg/constant/constant.go | 4 ++ 5 files changed, 60 insertions(+), 72 deletions(-) diff --git a/internal/service/client.go b/internal/service/client.go index 1cf23896..e1df85d4 100644 --- a/internal/service/client.go +++ b/internal/service/client.go @@ -49,6 +49,14 @@ func grpcClient(ctx *context.Context) (*grpc.ClientConn, *context.Context, error opts = append(opts, grpc.WithTransportCredentials(cred)) } + /*opts = append(opts, grpc.WithKeepaliveParams( + keepalive.ClientParameters{ + Time: 10 * time.Second, // Ping the server if idle for 10 seconds + Timeout: 15 * time.Second, // Wait 5 seconds for a ping ack before considering the connection dead + PermitWithoutStream: true, // Allow keepalive pings even with no active streams + }, + ))*/ + conn, err := grpc.NewClient(appConfig.BackendAddress, opts...) if err != nil { diff --git a/internal/service/component.go b/internal/service/component.go index 6bfc533c..e5c1059e 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "time" "github.com/briandowns/spinner" @@ -44,7 +45,7 @@ func (e *Component) OperateComponent(ctx *context.Context, request *serviceProto return message } - return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) } func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.OperateServiceRequest, stream serviceProto.ServiceService_OperateServiceClient) (serviceProto.ServiceService_OperateServiceClient, error) { @@ -52,12 +53,15 @@ func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx return nil, err } - newStream, err := client.OperateService(*requestCtx, request) - if err != nil { - return nil, err + for retries := 0; retries < constant.MaxConnectRetries; retries++ { + newStream, err := client.OperateService(*requestCtx, request) + if err == nil { + return newStream, nil + } + time.Sleep(constant.ConnectionRetryTimeout) } - return newStream, nil + return nil, fmt.Errorf(constant.FailedRetryMessage) } diff --git a/internal/service/service.go b/internal/service/service.go index bd78613f..54d7db63 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "time" "github.com/briandowns/spinner" "github.com/dream11/odin/pkg/constant" @@ -50,7 +51,7 @@ func (e *Service) DeployService(ctx *context.Context, request *serviceProto.Depl return message } - return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) } func logFailedComponentMessagesOnce(response *serviceProto.ServiceResponse) { @@ -176,7 +177,7 @@ func (e *Service) DeployReleasedService(ctx *context.Context, request *servicePr return message } - return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) } @@ -250,12 +251,16 @@ func (e *Service) OperateService(ctx *context.Context, request *serviceProto.Ope } generateResponse := func(response *serviceProto.OperateServiceResponse) string { - message := util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()}) - logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()}) + message := response.ServiceResponse.Message + message += fmt.Sprintf("\n Service %s %s", response.ServiceResponse.ServiceStatus.ServiceAction, response.ServiceResponse.ServiceStatus) + for _, compMessage := range response.ServiceResponse.ComponentsStatus { + message += fmt.Sprintf("\n Component %s %s %s", compMessage.ComponentName, compMessage.ComponentAction, compMessage.ComponentStatus) + } + logFailedComponentMessagesOnce(response.GetServiceResponse()) return message } - return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) } // ListService deploys service @@ -372,13 +377,15 @@ func reconnectDeployServiceStream(client serviceProto.ServiceServiceClient, requ return nil, err } - // Create a new stream connection - newStream, err := client.DeployService(*requestCtx, request) - if err != nil { - return nil, err - } + for retries := 0; retries < constant.MaxRetries; retries++ { + newStream, err := client.DeployService(*requestCtx, request) + if err == nil { + return newStream, nil + } - return newStream, nil + time.Sleep(constant.ConnectionRetryTimeout) + } + return nil, fmt.Errorf(constant.FailedRetryMessage) } func reconnectDeployReleasedServiceStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.DeployReleasedServiceRequest, stream serviceProto.ServiceService_DeployReleasedServiceClient) (serviceProto.ServiceService_DeployReleasedServiceClient, error) { @@ -387,11 +394,13 @@ func reconnectDeployReleasedServiceStream(client serviceProto.ServiceServiceClie return nil, err } - // Create a new stream connection - newStream, err := client.DeployReleasedService(*requestCtx, request) - if err != nil { - return nil, err - } + for retries := 0; retries < constant.MaxRetries; retries++ { + newStream, err := client.DeployReleasedService(*requestCtx, request) + if err == nil { + return newStream, nil + } - return newStream, nil + time.Sleep(constant.ConnectionRetryTimeout) + } + return nil, fmt.Errorf(constant.FailedRetryMessage) } diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go index d46115c4..249f7a3e 100644 --- a/internal/service/stream_handler.go +++ b/internal/service/stream_handler.go @@ -1,7 +1,6 @@ package service import ( - "context" "fmt" "io" @@ -23,68 +22,32 @@ type ReconnectFunc[S any] func() (S, error) type GenerateResponse[R any] func(response R) string // handleStreamResponse handles the stream response, retries on errors, and updates the spinner -func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, requestCtx *context.Context, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { +func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { var message string var retries = 0 - errorChan := make(chan error) - responseChan := make(chan R) - - go func(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - default: - response, err := stream.Recv() - if err != nil { - errorChan <- err - } - responseChan <- response - } - } - }(*requestCtx) - for { - recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout) - - select { - case <-recvCtx.Done(): + response, err := stream.Recv() + if err != nil { spinnerInstance.Stop() - cancel() - if !util.CanPerformRetry(retries, constant.MaxRetries) { - return nil - } - var er error - stream, er = reconnect() - if er != nil { + if err == io.EOF { + log.Info(message) return nil } - retries++ - case err := <-errorChan: - spinnerInstance.Stop() - cancel() if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { - if err != io.EOF { - return err - } else if err == io.EOF { - log.Info(message) - } - return nil + return err } + retries++ stream, err = reconnect() if err != nil { return nil } - retries++ - case response := <-responseChan: + spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) + spinnerInstance.Start() + } else { spinnerInstance.Stop() - cancel() - if !util.IsNil(response) { - message = generateResponse(response) - spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) - spinnerInstance.Start() - retries = 0 - } + message = generateResponse(response) + spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) + spinnerInstance.Start() } } } diff --git a/pkg/constant/constant.go b/pkg/constant/constant.go index 97b1a92b..a896546e 100644 --- a/pkg/constant/constant.go +++ b/pkg/constant/constant.go @@ -32,9 +32,13 @@ const ( ConsentMessageTemplate = "\nYou are executing the above command on a restricted environment. Are you sure? Enter \033[1m%s\033[0m to continue:" // InitiatingRetryMessage is the message shown when retrying InitiatingRetryMessage = "Unable to reach Odin backend." + //FailedRetryMessage is the message shown when client fails to connect to backend after repeated attempts + FailedRetryMessage = "failed to connect to odin backend" MaxRetries = 5 + MaxConnectRetries = 10 Timeout = 20 * time.Second + ConnectionRetryTimeout= 5 * time.Second MaxRetriesReached = "Max retries reached. Exiting...\nPlease check:\n- Your internet connection: \n- VPN connected properly" DescribeEnv = "\n- Please use describe env command to check the status.\n \u001B[1m odin describe env --name %s\u001B[0m " RetryingMessage = "Retrying ... (%d/%d)" From 32956071cda0091ffc4bd384620e72d9a4cad8d7 Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Wed, 9 Apr 2025 20:03:02 +0530 Subject: [PATCH 4/8] removed context from few rpc calss --- internal/service/client.go | 12 +++++++----- internal/service/component.go | 4 +--- internal/service/environment.go | 4 +--- 3 files changed, 9 insertions(+), 11 deletions(-) diff --git a/internal/service/client.go b/internal/service/client.go index e1df85d4..80db56cc 100644 --- a/internal/service/client.go +++ b/internal/service/client.go @@ -4,7 +4,9 @@ import ( "context" "crypto/tls" "fmt" + "google.golang.org/grpc/keepalive" "strings" + "time" "github.com/dream11/odin/pkg/config" "github.com/dream11/odin/pkg/constant" @@ -49,13 +51,13 @@ func grpcClient(ctx *context.Context) (*grpc.ClientConn, *context.Context, error opts = append(opts, grpc.WithTransportCredentials(cred)) } - /*opts = append(opts, grpc.WithKeepaliveParams( + opts = append(opts, grpc.WithKeepaliveParams( keepalive.ClientParameters{ - Time: 10 * time.Second, // Ping the server if idle for 10 seconds - Timeout: 15 * time.Second, // Wait 5 seconds for a ping ack before considering the connection dead - PermitWithoutStream: true, // Allow keepalive pings even with no active streams + Time: 10 * time.Second, + Timeout: 20 * time.Second, + PermitWithoutStream: true, }, - ))*/ + )) conn, err := grpc.NewClient(appConfig.BackendAddress, opts...) diff --git a/internal/service/component.go b/internal/service/component.go index e5c1059e..d08e86b2 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -101,10 +101,8 @@ func (e *Component) DescribeComponentType(ctx *context.Context, request *compone func (e *Component) CompareOperationChanges(ctx *context.Context, request *serviceProto.OperateComponentDiffRequest) (*serviceProto.OperateComponentDiffResponse, error) { for retries := 0; retries < constant.MaxRetries; retries++ { - ctxWithTimeout, cancel := context.WithTimeout(*ctx, constant.Timeout) - defer cancel() - conn, requestCtx, err := grpcClient(&ctxWithTimeout) + conn, requestCtx, err := grpcClient(ctx) if err != nil { return nil, err } diff --git a/internal/service/environment.go b/internal/service/environment.go index 6ad40fe2..5caef7cb 100644 --- a/internal/service/environment.go +++ b/internal/service/environment.go @@ -186,10 +186,8 @@ func (e *Environment) EnvironmentStatus(ctx *context.Context, request *environme func (e *Environment) IsStrictEnvironment(ctx *context.Context, request *environment.IsStrictEnvironmentRequest) (*environment.IsStrictEnvironmentResponse, error) { for retries := 0; retries < constant.MaxRetries; retries++ { - ctxWithTimeout, cancel := context.WithTimeout(*ctx, constant.Timeout) - defer cancel() - conn, requestCtx, err := grpcClient(&ctxWithTimeout) + conn, requestCtx, err := grpcClient(ctx) if err != nil { return nil, err } From c9a6b2a4efef9646f1676a7662adb4bfc2897655 Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Thu, 10 Apr 2025 13:29:12 +0530 Subject: [PATCH 5/8] optimized code --- internal/service/client.go | 2 +- internal/service/component.go | 11 +++----- internal/service/service.go | 29 ++++++++------------ internal/service/stream_handler.go | 43 +++++++++++++++++++++++++----- pkg/constant/constant.go | 14 +++++----- 5 files changed, 60 insertions(+), 39 deletions(-) diff --git a/internal/service/client.go b/internal/service/client.go index 80db56cc..ab68fc2b 100644 --- a/internal/service/client.go +++ b/internal/service/client.go @@ -4,7 +4,6 @@ import ( "context" "crypto/tls" "fmt" - "google.golang.org/grpc/keepalive" "strings" "time" @@ -15,6 +14,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" "google.golang.org/grpc/metadata" ) diff --git a/internal/service/component.go b/internal/service/component.go index d08e86b2..1682c549 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -45,7 +45,7 @@ func (e *Component) OperateComponent(ctx *context.Context, request *serviceProto return message } - return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, reconnect, generateResponse) } func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.OperateServiceRequest, stream serviceProto.ServiceService_OperateServiceClient) (serviceProto.ServiceService_OperateServiceClient, error) { @@ -53,12 +53,9 @@ func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx return nil, err } - for retries := 0; retries < constant.MaxConnectRetries; retries++ { - newStream, err := client.OperateService(*requestCtx, request) - if err == nil { - return newStream, nil - } - time.Sleep(constant.ConnectionRetryTimeout) + newStream, err := client.OperateService(*requestCtx, request) + if err == nil { + return newStream, nil } return nil, fmt.Errorf(constant.FailedRetryMessage) diff --git a/internal/service/service.go b/internal/service/service.go index 54d7db63..650e6ac6 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "io" - "time" "github.com/briandowns/spinner" "github.com/dream11/odin/pkg/constant" @@ -51,7 +50,7 @@ func (e *Service) DeployService(ctx *context.Context, request *serviceProto.Depl return message } - return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, reconnect, generateResponse) } func logFailedComponentMessagesOnce(response *serviceProto.ServiceResponse) { @@ -177,7 +176,7 @@ func (e *Service) DeployReleasedService(ctx *context.Context, request *servicePr return message } - return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, reconnect, generateResponse) } @@ -260,7 +259,7 @@ func (e *Service) OperateService(ctx *context.Context, request *serviceProto.Ope return message } - return handleStreamResponse(stream, spinnerInstance, reconnect, generateResponse) + return handleStreamResponse(stream, reconnect, generateResponse) } // ListService deploys service @@ -377,14 +376,11 @@ func reconnectDeployServiceStream(client serviceProto.ServiceServiceClient, requ return nil, err } - for retries := 0; retries < constant.MaxRetries; retries++ { - newStream, err := client.DeployService(*requestCtx, request) - if err == nil { - return newStream, nil - } - - time.Sleep(constant.ConnectionRetryTimeout) + newStream, err := client.DeployService(*requestCtx, request) + if err == nil { + return newStream, nil } + return nil, fmt.Errorf(constant.FailedRetryMessage) } func reconnectDeployReleasedServiceStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.DeployReleasedServiceRequest, stream serviceProto.ServiceService_DeployReleasedServiceClient) (serviceProto.ServiceService_DeployReleasedServiceClient, error) { @@ -394,13 +390,10 @@ func reconnectDeployReleasedServiceStream(client serviceProto.ServiceServiceClie return nil, err } - for retries := 0; retries < constant.MaxRetries; retries++ { - newStream, err := client.DeployReleasedService(*requestCtx, request) - if err == nil { - return newStream, nil - } - - time.Sleep(constant.ConnectionRetryTimeout) + newStream, err := client.DeployReleasedService(*requestCtx, request) + if err == nil { + return newStream, nil } + return nil, fmt.Errorf(constant.FailedRetryMessage) } diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go index 249f7a3e..1622fa5b 100644 --- a/internal/service/stream_handler.go +++ b/internal/service/stream_handler.go @@ -3,6 +3,7 @@ package service import ( "fmt" "io" + "time" "github.com/briandowns/spinner" "github.com/dream11/odin/pkg/constant" @@ -22,9 +23,14 @@ type ReconnectFunc[S any] func() (S, error) type GenerateResponse[R any] func(response R) string // handleStreamResponse handles the stream response, retries on errors, and updates the spinner -func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { +func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error { + spinnerInstance := spinner.New(spinner.CharSets[constant.SpinnerType], constant.SpinnerDelay) + err := spinnerInstance.Color(constant.SpinnerColor, constant.SpinnerStyle) + if err != nil { + return err + } + var message string - var retries = 0 for { response, err := stream.Recv() if err != nil { @@ -33,13 +39,12 @@ func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, spinner log.Info(message) return nil } - if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) { + if !util.IsRetryable(err) { return err } - retries++ - stream, err = reconnect() + stream, err = retryWithReconnect(reconnect, constant.MaxConnectRetries, constant.ConnectionRetryTimeout) if err != nil { - return nil + return err } spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) spinnerInstance.Start() @@ -51,3 +56,29 @@ func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, spinner } } } + +func retryWithReconnect[S any](reconnect ReconnectFunc[S], maxRetries int, retryTimeout time.Duration) (S, error) { + var stream S + var err error + + for retries := 0; retries < maxRetries; retries++ { + + if retries == 0 { + log.Warnf(constant.InitiatingRetryMessage) + } + + if retries < maxRetries { + log.Infof(constant.RetryingMessage, retries+1, maxRetries) + } + + stream, err = reconnect() + if err == nil { + return stream, nil + } + + time.Sleep(retryTimeout) + } + log.Errorf("%s", constant.MaxRetriesReached) + + return stream, err +} diff --git a/pkg/constant/constant.go b/pkg/constant/constant.go index a896546e..a611d4b6 100644 --- a/pkg/constant/constant.go +++ b/pkg/constant/constant.go @@ -35,11 +35,11 @@ const ( //FailedRetryMessage is the message shown when client fails to connect to backend after repeated attempts FailedRetryMessage = "failed to connect to odin backend" - MaxRetries = 5 - MaxConnectRetries = 10 - Timeout = 20 * time.Second - ConnectionRetryTimeout= 5 * time.Second - MaxRetriesReached = "Max retries reached. Exiting...\nPlease check:\n- Your internet connection: \n- VPN connected properly" - DescribeEnv = "\n- Please use describe env command to check the status.\n \u001B[1m odin describe env --name %s\u001B[0m " - RetryingMessage = "Retrying ... (%d/%d)" + MaxRetries = 5 + MaxConnectRetries = 10 + Timeout = 20 * time.Second + ConnectionRetryTimeout = 5 * time.Second + MaxRetriesReached = "Max retries reached. Exiting...\nPlease check:\n- Your internet connection: \n- VPN connected properly" + DescribeEnv = "\n- Please use describe env command to check the status.\n \u001B[1m odin describe env --name %s\u001B[0m " + RetryingMessage = "Retrying ... (%d/%d)" ) From b2c6998cfe6a6a4bf2ea6d5af3697af368e3f149 Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Thu, 10 Apr 2025 13:30:33 +0530 Subject: [PATCH 6/8] optimized code --- internal/service/stream_handler.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go index 1622fa5b..fb96be83 100644 --- a/internal/service/stream_handler.go +++ b/internal/service/stream_handler.go @@ -42,7 +42,7 @@ func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, reconne if !util.IsRetryable(err) { return err } - stream, err = retryWithReconnect(reconnect, constant.MaxConnectRetries, constant.ConnectionRetryTimeout) + stream, err = reconnectWithRetry(reconnect, constant.MaxConnectRetries, constant.ConnectionRetryTimeout) if err != nil { return err } @@ -57,7 +57,7 @@ func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, reconne } } -func retryWithReconnect[S any](reconnect ReconnectFunc[S], maxRetries int, retryTimeout time.Duration) (S, error) { +func reconnectWithRetry[S any](reconnect ReconnectFunc[S], maxRetries int, retryTimeout time.Duration) (S, error) { var stream S var err error From f96351b62b071ca24080fbcb750152ba661f19a0 Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Thu, 10 Apr 2025 13:34:02 +0530 Subject: [PATCH 7/8] optimized code --- internal/service/component.go | 7 ------- internal/service/service.go | 15 --------------- 2 files changed, 22 deletions(-) diff --git a/internal/service/component.go b/internal/service/component.go index 1682c549..65a4dc02 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -5,7 +5,6 @@ import ( "fmt" "time" - "github.com/briandowns/spinner" "github.com/dream11/odin/pkg/constant" "github.com/dream11/odin/pkg/util" component "github.com/dream11/odin/proto/gen/go/dream11/od/component/v1" @@ -29,12 +28,6 @@ func (e *Component) OperateComponent(ctx *context.Context, request *serviceProto } log.Info("Starting component operation...") - spinnerInstance := spinner.New(spinner.CharSets[constant.SpinnerType], constant.SpinnerDelay) - err = spinnerInstance.Color(constant.SpinnerColor, constant.SpinnerStyle) - if err != nil { - return err - } - reconnect := func() (serviceProto.ServiceService_OperateServiceClient, error) { return reconnectOperateStream(client, requestCtx, request, stream) } diff --git a/internal/service/service.go b/internal/service/service.go index 650e6ac6..a9cab960 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -34,11 +34,6 @@ func (e *Service) DeployService(ctx *context.Context, request *serviceProto.Depl } log.Info("Deploying Service...") - spinnerInstance := spinner.New(spinner.CharSets[constant.SpinnerType], constant.SpinnerDelay) - err = spinnerInstance.Color(constant.SpinnerColor, constant.SpinnerStyle) - if err != nil { - return err - } reconnect := func() (serviceProto.ServiceService_DeployServiceClient, error) { return reconnectDeployServiceStream(client, requestCtx, request, stream) @@ -157,11 +152,6 @@ func (e *Service) DeployReleasedService(ctx *context.Context, request *servicePr } log.Info("Deploying Service...") - spinnerInstance := spinner.New(spinner.CharSets[constant.SpinnerType], constant.SpinnerDelay) - err = spinnerInstance.Color(constant.SpinnerColor, constant.SpinnerStyle) - if err != nil { - return err - } reconnect := func() (serviceProto.ServiceService_DeployReleasedServiceClient, error) { return reconnectDeployReleasedServiceStream(client, requestCtx, request, stream) } @@ -239,11 +229,6 @@ func (e *Service) OperateService(ctx *context.Context, request *serviceProto.Ope } log.Info("Starting service operation...") - spinnerInstance := spinner.New(spinner.CharSets[constant.SpinnerType], constant.SpinnerDelay) - err = spinnerInstance.Color(constant.SpinnerColor, constant.SpinnerStyle) - if err != nil { - return err - } reconnect := func() (serviceProto.ServiceService_OperateServiceClient, error) { return reconnectOperateStream(client, requestCtx, request, stream) From 330c40c51d324af67bd33bf32d5fd946a9c53671 Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Thu, 10 Apr 2025 14:53:36 +0530 Subject: [PATCH 8/8] minor improvements --- internal/pkg/retry/retry.go | 59 +++++++++++++++++++++++++++++++++ internal/service/component.go | 6 ++-- internal/service/environment.go | 6 ++-- pkg/constant/constant.go | 7 ++-- pkg/util/util.go | 34 ------------------- 5 files changed, 68 insertions(+), 44 deletions(-) create mode 100644 internal/pkg/retry/retry.go diff --git a/internal/pkg/retry/retry.go b/internal/pkg/retry/retry.go new file mode 100644 index 00000000..ff993d89 --- /dev/null +++ b/internal/pkg/retry/retry.go @@ -0,0 +1,59 @@ +package retry + +import ( + "time" + + log "github.com/sirupsen/logrus" +) + +// RetryConfig holds the configuration for retry behavior +type RetryConfig struct { + // MaxRetries is the maximum number of retry attempts + MaxRetries int + // RetryDelay is the delay between retry attempts + RetryDelay time.Duration + // RetryableError is a function that determines if an error is retryable + RetryableError func(error) bool +} + +// DefaultRetryConfig returns a default retry configuration +func DefaultRetryConfig() *RetryConfig { + return &RetryConfig{ + MaxRetries: 3, + RetryDelay: time.Second, + RetryableError: func(err error) bool { return true }, + } +} + +// Retry executes the given function and retries if it fails based on the provided configuration +func Retry[T any](fn func() (T, error), config *RetryConfig) (T, error) { + var result T + var err error + + if config == nil { + config = DefaultRetryConfig() + } + + for attempt := 0; attempt <= config.MaxRetries; attempt++ { + if attempt > 0 { + log.Infof("Retry attempt %d/%d", attempt, config.MaxRetries) + time.Sleep(config.RetryDelay) + } + + result, err = fn() + if err == nil { + return result, nil + } + + if !config.RetryableError(err) { + return result, err + } + + if attempt == config.MaxRetries { + log.Errorf("Max retries (%d) reached. Last error: %v", config.MaxRetries, err) + return result, err + } + } + + return result, err +} diff --git a/internal/service/component.go b/internal/service/component.go index 65a4dc02..6865dd91 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -90,7 +90,7 @@ func (e *Component) DescribeComponentType(ctx *context.Context, request *compone // CompareOperationChanges compares the operation changes func (e *Component) CompareOperationChanges(ctx *context.Context, request *serviceProto.OperateComponentDiffRequest) (*serviceProto.OperateComponentDiffResponse, error) { - for retries := 0; retries < constant.MaxRetries; retries++ { + for retries := 0; retries < constant.MaxConnectRetries; retries++ { conn, requestCtx, err := grpcClient(ctx) if err != nil { @@ -106,11 +106,11 @@ func (e *Component) CompareOperationChanges(ctx *context.Context, request *servi if !util.IsRetryable(err) { return nil, err } - time.Sleep(constant.Timeout) + time.Sleep(constant.ConnectionRetryTimeout) if retries == 0 { log.Warnf(constant.InitiatingRetryMessage) } - log.Infof(constant.RetryingMessage, retries+1, constant.MaxRetries) + log.Infof(constant.RetryingMessage, retries+1, constant.MaxConnectRetries) } log.Fatalf(constant.MaxRetriesReached) diff --git a/internal/service/environment.go b/internal/service/environment.go index 5caef7cb..e3ab3435 100644 --- a/internal/service/environment.go +++ b/internal/service/environment.go @@ -185,7 +185,7 @@ func (e *Environment) EnvironmentStatus(ctx *context.Context, request *environme // IsStrictEnvironment checks if the given environment is a strict environment func (e *Environment) IsStrictEnvironment(ctx *context.Context, request *environment.IsStrictEnvironmentRequest) (*environment.IsStrictEnvironmentResponse, error) { - for retries := 0; retries < constant.MaxRetries; retries++ { + for retries := 0; retries < constant.MaxConnectRetries; retries++ { conn, requestCtx, err := grpcClient(ctx) if err != nil { @@ -202,11 +202,11 @@ func (e *Environment) IsStrictEnvironment(ctx *context.Context, request *environ log.Errorf("TraceID: %s", (*requestCtx).Value(constant.TraceIDKey)) return nil, err } - time.Sleep(constant.Timeout) + time.Sleep(constant.ConnectionRetryTimeout) if retries == 0 { log.Warnf(constant.InitiatingRetryMessage) } - log.Infof(constant.RetryingMessage, retries+1, constant.MaxRetries) + log.Infof(constant.RetryingMessage, retries+1, constant.MaxConnectRetries) } log.Fatalf(constant.MaxRetriesReached) diff --git a/pkg/constant/constant.go b/pkg/constant/constant.go index a611d4b6..78d347a8 100644 --- a/pkg/constant/constant.go +++ b/pkg/constant/constant.go @@ -34,10 +34,9 @@ const ( InitiatingRetryMessage = "Unable to reach Odin backend." //FailedRetryMessage is the message shown when client fails to connect to backend after repeated attempts FailedRetryMessage = "failed to connect to odin backend" - - MaxRetries = 5 - MaxConnectRetries = 10 - Timeout = 20 * time.Second + // MaxConnectRetries is the maximum number of connection retries + MaxConnectRetries = 10 + // ConnectionRetryTimeout is the timeout for connection retries ConnectionRetryTimeout = 5 * time.Second MaxRetriesReached = "Max retries reached. Exiting...\nPlease check:\n- Your internet connection: \n- VPN connected properly" DescribeEnv = "\n- Please use describe env command to check the status.\n \u001B[1m odin describe env --name %s\u001B[0m " diff --git a/pkg/util/util.go b/pkg/util/util.go index c6ba1c31..e9679d82 100644 --- a/pkg/util/util.go +++ b/pkg/util/util.go @@ -8,13 +8,11 @@ import ( "io" "net" "os" - "reflect" "strconv" "strings" "time" "github.com/dream11/odin/internal/ui" - "github.com/dream11/odin/pkg/constant" v1 "github.com/dream11/odin/proto/gen/go/dream11/od/service/v1" "github.com/google/uuid" "github.com/olekukonko/tablewriter" @@ -193,35 +191,3 @@ func IsRetryable(err error) bool { st, ok := status.FromError(err) return ok && (st.Code() == codes.Unavailable || (st.Code() == codes.Internal && strings.Contains(st.Message(), "RST_STREAM"))) } - -// CanPerformRetry checks if the operation can be retried -func CanPerformRetry( - retries int, - maxRetries int, -) bool { - if retries == maxRetries { - log.Errorf("%s", constant.MaxRetriesReached) - return false - } - - if retries == 0 { - log.Warnf(constant.InitiatingRetryMessage) - } - - if retries < constant.MaxRetries { - log.Infof(constant.RetryingMessage, retries+1, constant.MaxRetries) - } - - return true -} - -// IsNil checks if a value is nil -func IsNil[T any](v T) bool { - val := reflect.ValueOf(v) - switch val.Kind() { - case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: - return val.IsNil() - default: - return false - } -}