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/client.go b/internal/service/client.go index 1cf23896..ab68fc2b 100644 --- a/internal/service/client.go +++ b/internal/service/client.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "fmt" "strings" + "time" "github.com/dream11/odin/pkg/config" "github.com/dream11/odin/pkg/constant" @@ -13,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" ) @@ -49,6 +51,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, + Timeout: 20 * time.Second, + PermitWithoutStream: true, + }, + )) + conn, err := grpc.NewClient(appConfig.BackendAddress, opts...) if err != nil { diff --git a/internal/service/component.go b/internal/service/component.go index 95c2f327..6865dd91 100644 --- a/internal/service/component.go +++ b/internal/service/component.go @@ -3,10 +3,8 @@ package service import ( "context" "fmt" - "io" "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" @@ -30,75 +28,17 @@ 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) } - 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 - } - } + generateResponse := func(response *serviceProto.OperateServiceResponse) string { + message := util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()}) + logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()}) + return message } + + return handleStreamResponse(stream, reconnect, generateResponse) } func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.OperateServiceRequest, stream serviceProto.ServiceService_OperateServiceClient) (serviceProto.ServiceService_OperateServiceClient, error) { @@ -107,11 +47,11 @@ func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx } newStream, err := client.OperateService(*requestCtx, request) - if err != nil { - return nil, err + if err == nil { + return newStream, nil } - return newStream, nil + return nil, fmt.Errorf(constant.FailedRetryMessage) } @@ -150,11 +90,9 @@ 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++ { - ctxWithTimeout, cancel := context.WithTimeout(*ctx, constant.Timeout) - defer cancel() + for retries := 0; retries < constant.MaxConnectRetries; retries++ { - conn, requestCtx, err := grpcClient(&ctxWithTimeout) + conn, requestCtx, err := grpcClient(ctx) if err != nil { return nil, err } @@ -168,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 6ad40fe2..e3ab3435 100644 --- a/internal/service/environment.go +++ b/internal/service/environment.go @@ -185,11 +185,9 @@ 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++ { - ctxWithTimeout, cancel := context.WithTimeout(*ctx, constant.Timeout) - defer cancel() + for retries := 0; retries < constant.MaxConnectRetries; retries++ { - conn, requestCtx, err := grpcClient(&ctxWithTimeout) + conn, requestCtx, err := grpcClient(ctx) if err != nil { return nil, err } @@ -204,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/internal/service/service.go b/internal/service/service.go index 26de94b7..a9cab960 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -34,75 +34,18 @@ 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 - } - - 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 handleStreamResponse(stream, reconnect, generateResponse) } func logFailedComponentMessagesOnce(response *serviceProto.ServiceResponse) { @@ -209,79 +152,22 @@ 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) } - 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 handleStreamResponse(stream, reconnect, generateResponse) + } // UndeployService undeploy service @@ -343,74 +229,22 @@ 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 - } - 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 := 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, reconnect, generateResponse) } // ListService deploys service @@ -527,13 +361,12 @@ 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 + if err == nil { + return newStream, 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) { @@ -542,11 +375,10 @@ 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 + if err == nil { + return newStream, nil } - return newStream, nil + return nil, fmt.Errorf(constant.FailedRetryMessage) } diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go new file mode 100644 index 00000000..fb96be83 --- /dev/null +++ b/internal/service/stream_handler.go @@ -0,0 +1,84 @@ +package service + +import ( + "fmt" + "io" + "time" + + "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 + +// handleStreamResponse handles the stream response, retries on errors, and updates the spinner +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 + for { + response, err := stream.Recv() + if err != nil { + spinnerInstance.Stop() + if err == io.EOF { + log.Info(message) + return nil + } + if !util.IsRetryable(err) { + return err + } + stream, err = reconnectWithRetry(reconnect, constant.MaxConnectRetries, constant.ConnectionRetryTimeout) + if err != nil { + return err + } + spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) + spinnerInstance.Start() + } else { + spinnerInstance.Stop() + message = generateResponse(response) + spinnerInstance.Prefix = fmt.Sprintf(" %s ", message) + spinnerInstance.Start() + } + } +} + +func reconnectWithRetry[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 97b1a92b..78d347a8 100644 --- a/pkg/constant/constant.go +++ b/pkg/constant/constant.go @@ -32,10 +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." - - MaxRetries = 5 - Timeout = 20 * 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)" + //FailedRetryMessage is the message shown when client fails to connect to backend after repeated attempts + FailedRetryMessage = "failed to connect to odin backend" + // 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 " + RetryingMessage = "Retrying ... (%d/%d)" ) diff --git a/pkg/util/util.go b/pkg/util/util.go index 59b8b85c..e9679d82 100644 --- a/pkg/util/util.go +++ b/pkg/util/util.go @@ -13,7 +13,6 @@ import ( "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" @@ -192,24 +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 -}