From e548acc71208820c5c465b684ce1f4f2c15a87cf Mon Sep 17 00:00:00 2001 From: nikhil-024 Date: Tue, 8 Apr 2025 17:03:52 +0530 Subject: [PATCH 1/2] 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/2] 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)