Feat/retry stream generic v2

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
+}

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 {

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)

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)