panic recovered go-routines

Author: prkhrkat

api/restHandler/BatchOperationRestHandler.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"net/http"
 
+	"github.com/devtron-labs/common-lib/async"
 	"github.com/devtron-labs/devtron/api/restHandler/common"
 	"github.com/devtron-labs/devtron/pkg/apis/devtron/v1"
 	"github.com/devtron-labs/devtron/pkg/apis/devtron/v1/validation"
@@ -44,17 +45,19 @@ type BatchOperationRestHandlerImpl struct {
 	teamService     team.TeamService
 	logger          *zap.SugaredLogger
 	enforcerUtil    rbac.EnforcerUtil
+	asyncRunnable   *async.Runnable
 }
 
 func NewBatchOperationRestHandlerImpl(userAuthService user.UserService, enforcer casbin.Enforcer, workflowAction batch.WorkflowAction,
-	teamService team.TeamService, logger *zap.SugaredLogger, enforcerUtil rbac.EnforcerUtil) *BatchOperationRestHandlerImpl {
+	teamService team.TeamService, logger *zap.SugaredLogger, enforcerUtil rbac.EnforcerUtil, asyncRunnable *async.Runnable) *BatchOperationRestHandlerImpl {
 	return &BatchOperationRestHandlerImpl{
 		userAuthService: userAuthService,
 		enforcer:        enforcer,
 		workflowAction:  workflowAction,
 		teamService:     teamService,
 		logger:          logger,
 		enforcerUtil:    enforcerUtil,
+		asyncRunnable:   asyncRunnable,
 	}
 }
 
@@ -104,13 +107,14 @@ func (handler BatchOperationRestHandlerImpl) Operate(w http.ResponseWriter, r *h
 
 		ctx, cancel := context.WithCancel(r.Context())
 		if cn, ok := w.(http.CloseNotifier); ok {
-			go func(done <-chan struct{}, closed <-chan bool) {
+			runnableFunc := func(done <-chan struct{}, closed <-chan bool) {
 				select {
 				case <-done:
 				case <-closed:
 					cancel()
 				}
-			}(ctx.Done(), cn.CloseNotify())
+			}
+			handler.asyncRunnable.Execute(func() { runnableFunc(ctx.Done(), cn.CloseNotify()) })
 		}
 		err = handler.workflowAction.Execute(&workflow, emptyProps, r.Context())
 		if err != nil {

pkg/argoApplication/ArgoApplicationServiceExtended.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	application2 "github.com/argoproj/argo-cd/v2/pkg/apiclient/application"
 	"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
+	"github.com/devtron-labs/common-lib/async"
 	"github.com/devtron-labs/common-lib/utils/k8s"
 	openapi "github.com/devtron-labs/devtron/api/helm-app/openapiClient"
 	"github.com/devtron-labs/devtron/client/argocdServer"
@@ -46,6 +47,7 @@ type ArgoApplicationServiceExtendedImpl struct {
 	argoApplicationReadService read.ArgoApplicationReadService
 	clusterService             cluster.ClusterService
 	acdClientWrapper           argocdServer.ArgoClientWrapperService
+	asyncRunnable              *async.Runnable
 }
 
 func NewArgoApplicationServiceExtendedServiceImpl(
@@ -54,13 +56,15 @@ func NewArgoApplicationServiceExtendedServiceImpl(
 	acdClientWrapper argocdServer.ArgoClientWrapperService,
 	argoApplicationReadService read.ArgoApplicationReadService,
 	clusterService cluster.ClusterService,
+	asyncRunnable *async.Runnable,
 ) *ArgoApplicationServiceExtendedImpl {
 	return &ArgoApplicationServiceExtendedImpl{
 		aCDAuthConfig:              aCDAuthConfig,
 		clusterService:             clusterService,
 		argoApplicationReadService: argoApplicationReadService,
 		acdClientWrapper:           acdClientWrapper,
 		ArgoApplicationServiceImpl: argoApplicationServiceImpl,
+		asyncRunnable:              asyncRunnable,
 	}
 }
 
@@ -328,7 +332,7 @@ func (c *ArgoApplicationServiceExtendedImpl) parseResult(resp *v1alpha1.Applicat
 	for _, node := range queryNodes {
 		rQuery := transform(node, query.ApplicationName)
 		qCount++
-		go func(request application2.ApplicationResourceRequest) {
+		runnableFunc := func(request application2.ApplicationResourceRequest) {
 			ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
 			defer cancel()
 			startTime := time.Now()
@@ -343,7 +347,8 @@ func (c *ArgoApplicationServiceExtendedImpl) parseResult(resp *v1alpha1.Applicat
 			} else {
 				response <- argoApplication.Result{Response: nil, Error: fmt.Errorf("connection closed by client"), Request: &request}
 			}
-		}(*rQuery)
+		}
+		c.asyncRunnable.Execute(func() { runnableFunc(*rQuery) })
 	}
 
 	if qCount == 0 {

pkg/cluster/ClusterService.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/devtron-labs/common-lib/async"
 	informerBean "github.com/devtron-labs/common-lib/informer"
 	"github.com/devtron-labs/common-lib/utils/k8s/commonBean"
 	"github.com/devtron-labs/devtron/pkg/cluster/adapter"
@@ -97,6 +98,7 @@ type ClusterServiceImpl struct {
 	userRepository      repository3.UserRepository
 	roleGroupRepository repository3.RoleGroupRepository
 	clusterReadService  read.ClusterReadService
+	asyncRunnable       *async.Runnable
 }
 
 func NewClusterServiceImpl(repository repository.ClusterRepository, logger *zap.SugaredLogger,
@@ -105,7 +107,8 @@ func NewClusterServiceImpl(repository repository.ClusterRepository, logger *zap.
 	roleGroupRepository repository3.RoleGroupRepository,
 	envVariables *globalUtil.EnvironmentVariables,
 	cronLogger *cronUtil.CronLoggerImpl,
-	clusterReadService read.ClusterReadService) (*ClusterServiceImpl, error) {
+	clusterReadService read.ClusterReadService,
+	asyncRunnable *async.Runnable) (*ClusterServiceImpl, error) {
 	clusterService := &ClusterServiceImpl{
 		clusterRepository:   repository,
 		logger:              logger,
@@ -115,6 +118,7 @@ func NewClusterServiceImpl(repository repository.ClusterRepository, logger *zap.
 		userRepository:      userRepository,
 		roleGroupRepository: roleGroupRepository,
 		clusterReadService:  clusterReadService,
+		asyncRunnable:       asyncRunnable,
 	}
 	// initialise cron
 	newCron := cron.New(cron.WithChain(cron.Recover(cronLogger)))
@@ -782,7 +786,7 @@ func (impl *ClusterServiceImpl) ConnectClustersInBatch(clusters []*bean.ClusterB
 			continue
 		}
 		wg.Add(1)
-		go func(idx int, cluster *bean.ClusterBean) {
+		runnableFunc := func(idx int, cluster *bean.ClusterBean) {
 			defer wg.Done()
 			clusterConfig := cluster.GetClusterConfig()
 			_, _, k8sClientSet, err := impl.K8sUtil.GetK8sConfigAndClients(clusterConfig)
@@ -796,7 +800,8 @@ func (impl *ClusterServiceImpl) ConnectClustersInBatch(clusters []*bean.ClusterB
 				id = idx
 			}
 			impl.GetAndUpdateConnectionStatusForOneCluster(k8sClientSet, id, respMap)
-		}(idx, cluster)
+		}
+		impl.asyncRunnable.Execute(func() { runnableFunc(idx, cluster) })
 	}
 
 	wg.Wait()

pkg/k8s/K8sCommonService.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/caarlos0/env"
+	"github.com/devtron-labs/common-lib/async"
 	"github.com/devtron-labs/common-lib/utils/k8s"
 	k8sCommonBean "github.com/devtron-labs/common-lib/utils/k8s/commonBean"
 	"github.com/devtron-labs/devtron/api/bean/AppView"
@@ -75,6 +76,7 @@ type K8sCommonServiceImpl struct {
 	K8sApplicationServiceConfig  *K8sApplicationServiceConfig
 	argoApplicationConfigService config.ArgoApplicationConfigService
 	ClusterReadService           read.ClusterReadService
+	asyncRunnable                *async.Runnable
 }
 type K8sApplicationServiceConfig struct {
 	BatchSize        int `env:"BATCH_SIZE" envDefault:"5" description:"there is feature to get URL's of services/ingresses. so to extract those, we need to parse all the servcie and ingress objects of the application. this BATCH_SIZE flag controls the no of these objects get parsed in one go."`
@@ -83,7 +85,7 @@ type K8sApplicationServiceConfig struct {
 
 func NewK8sCommonServiceImpl(Logger *zap.SugaredLogger, k8sUtils *k8s.K8sServiceImpl,
 	argoApplicationConfigService config.ArgoApplicationConfigService,
-	ClusterReadService read.ClusterReadService) *K8sCommonServiceImpl {
+	ClusterReadService read.ClusterReadService, asyncRunnable *async.Runnable) *K8sCommonServiceImpl {
 	cfg := &K8sApplicationServiceConfig{}
 	err := env.Parse(cfg)
 	if err != nil {
@@ -95,6 +97,7 @@ func NewK8sCommonServiceImpl(Logger *zap.SugaredLogger, k8sUtils *k8s.K8sService
 		K8sApplicationServiceConfig:  cfg,
 		argoApplicationConfigService: argoApplicationConfigService,
 		ClusterReadService:           ClusterReadService,
+		asyncRunnable:                asyncRunnable,
 	}
 }
 
@@ -295,15 +298,18 @@ func (impl *K8sCommonServiceImpl) GetManifestsByBatch(ctx context.Context, reque
 	var res []bean5.BatchResourceResponse
 	ctx, cancel := context.WithTimeout(ctx, time.Duration(impl.K8sApplicationServiceConfig.TimeOutInSeconds)*time.Second)
 	defer cancel()
-	go func() {
+
+	runnableFunc := func() {
 		ans := impl.getManifestsByBatch(ctx, requests)
 		select {
 		case <-ctx.Done():
 			return
 		default:
 			ch <- ans
 		}
-	}()
+	}
+	impl.asyncRunnable.Execute(runnableFunc)
+
 	select {
 	case ans := <-ch:
 		res = ans
@@ -398,7 +404,7 @@ func (impl *K8sCommonServiceImpl) getManifestsByBatch(ctx context.Context, reque
 		var wg sync.WaitGroup
 		for j := 0; j < batchSize; j++ {
 			wg.Add(1)
-			go func(j int) {
+			runnableFunc := func(j int) {
 				resp := bean5.BatchResourceResponse{}
 				response, err := impl.GetResource(ctx, &requests[i+j])
 				if response != nil {
@@ -407,7 +413,8 @@ func (impl *K8sCommonServiceImpl) getManifestsByBatch(ctx context.Context, reque
 				resp.Err = err
 				res[i+j] = resp
 				wg.Done()
-			}(j)
+			}
+			impl.asyncRunnable.Execute(func() { runnableFunc(j) })
 		}
 		wg.Wait()
 		i += batchSize

pkg/workflow/dag/WorkflowDagExecutor.go

@@ -1075,15 +1075,16 @@ func (impl *WorkflowDagExecutorImpl) HandleCiSuccessEvent(triggerContext trigger
 		for j := 0; j < batchSize; j++ {
 			wg.Add(1)
 			index := i + j
-			go func(index int) {
+			runnableFunc := func(index int) {
 				defer wg.Done()
 				ciArtifact := ciArtifactArr[index]
 				// handle individual CiArtifact success event
 				err = impl.handleCiSuccessEvent(triggerContext, ciArtifact, async, request.UserId)
 				if err != nil {
 					impl.logger.Errorw("error on handle  ci success event", "ciArtifactId", ciArtifact.Id, "err", err)
 				}
-			}(index)
+			}
+			impl.asyncRunnable.Execute(func() { runnableFunc(index) })
 		}
 		wg.Wait()
 		i += batchSize