panic recovered go-routines

Author: prkhrkat

pkg/app/AppService.go

@@ -20,6 +20,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/url"
+	"strconv"
+	"time"
+
 	health2 "github.com/argoproj/gitops-engine/pkg/health"
 	argoApplication "github.com/devtron-labs/devtron/client/argocdServer/bean"
 	"github.com/devtron-labs/devtron/internal/sql/models"
@@ -40,12 +44,10 @@ import (
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate/read"
 	bean4 "github.com/devtron-labs/devtron/pkg/deployment/trigger/devtronApps/bean"
 	"github.com/devtron-labs/devtron/pkg/workflow/cd"
-	"net/url"
-	"strconv"
-	"time"
 
 	"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
 	"github.com/caarlos0/env"
+	"github.com/devtron-labs/common-lib/async"
 	k8sCommonBean "github.com/devtron-labs/common-lib/utils/k8s/commonBean"
 	"github.com/devtron-labs/common-lib/utils/k8s/health"
 	"github.com/devtron-labs/devtron/api/bean"
@@ -124,6 +126,7 @@ type AppServiceImpl struct {
 	deploymentConfigService                common2.DeploymentConfigService
 	envConfigOverrideReadService           read.EnvConfigOverrideService
 	cdWorkflowRunnerService                cd.CdWorkflowRunnerService
+	asyncRunnable                          *async.Runnable
 }
 
 type AppService interface {
@@ -151,7 +154,7 @@ func NewAppService(
 	cdWorkflowRepository pipelineConfig.CdWorkflowRepository,
 	commonService commonService.CommonService,
 	chartTemplateService internalUtil.ChartTemplateService,
-	cdPipelineStatusTimelineRepo pipelineConfig.PipelineStatusTimelineRepository,
+	pipelineStatusTimelineRepo pipelineConfig.PipelineStatusTimelineRepository,
 	pipelineStatusTimelineResourcesService status2.PipelineStatusTimelineResourcesService,
 	pipelineStatusSyncDetailService status2.PipelineStatusSyncDetailService,
 	pipelineStatusTimelineService status2.PipelineStatusTimelineService,
@@ -164,7 +167,8 @@ func NewAppService(
 	appListingService AppListingService,
 	deploymentConfigService common2.DeploymentConfigService,
 	envConfigOverrideReadService read.EnvConfigOverrideService,
-	cdWorkflowRunnerService cd.CdWorkflowRunnerService) *AppServiceImpl {
+	cdWorkflowRunnerService cd.CdWorkflowRunnerService,
+	asyncRunnable *async.Runnable) *AppServiceImpl {
 	appServiceImpl := &AppServiceImpl{
 		mergeUtil:                              mergeUtil,
 		pipelineOverrideRepository:             pipelineOverrideRepository,
@@ -178,7 +182,7 @@ func NewAppService(
 		cdWorkflowRepository:                   cdWorkflowRepository,
 		commonService:                          commonService,
 		chartTemplateService:                   chartTemplateService,
-		pipelineStatusTimelineRepository:       cdPipelineStatusTimelineRepo,
+		pipelineStatusTimelineRepository:       pipelineStatusTimelineRepo,
 		pipelineStatusTimelineResourcesService: pipelineStatusTimelineResourcesService,
 		pipelineStatusSyncDetailService:        pipelineStatusSyncDetailService,
 		pipelineStatusTimelineService:          pipelineStatusTimelineService,
@@ -195,6 +199,7 @@ func NewAppService(
 		deploymentConfigService:                deploymentConfigService,
 		envConfigOverrideReadService:           envConfigOverrideReadService,
 		cdWorkflowRunnerService:                cdWorkflowRunnerService,
+		asyncRunnable:                          asyncRunnable,
 	}
 	return appServiceImpl
 }
@@ -320,7 +325,7 @@ func (impl *AppServiceImpl) UpdateDeploymentStatusForGitOpsPipelines(app *v1alph
 				}
 				if isSucceeded {
 					impl.logger.Infow("writing cd success event", "gitHash", gitHash, "pipelineOverride", pipelineOverride)
-					go impl.WriteCDSuccessEvent(cdPipeline.AppId, cdPipeline.EnvironmentId, pipelineOverride)
+					impl.asyncRunnable.Execute(func() { impl.WriteCDSuccessEvent(cdPipeline.AppId, cdPipeline.EnvironmentId, pipelineOverride) })
 				}
 			} else {
 				impl.logger.Debugw("event received for older triggered revision", "gitHash", gitHash)

pkg/clusterTerminalAccess/UserTerminalAccessService.go

@@ -21,11 +21,18 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/caarlos0/env/v6"
+	"github.com/devtron-labs/common-lib/async"
 	k8s2 "github.com/devtron-labs/common-lib/utils/k8s"
 	"github.com/devtron-labs/devtron/api/helm-app/service/bean"
 	"github.com/devtron-labs/devtron/internal/sql/models"
 	"github.com/devtron-labs/devtron/internal/sql/repository"
+	"github.com/devtron-labs/devtron/pkg/asyncProvider"
 	utils1 "github.com/devtron-labs/devtron/pkg/clusterTerminalAccess/clusterTerminalUtils"
 	"github.com/devtron-labs/devtron/pkg/k8s"
 	bean2 "github.com/devtron-labs/devtron/pkg/k8s/bean"
@@ -43,10 +50,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
 )
 
 type UserTerminalAccessService interface {
@@ -75,6 +78,7 @@ type UserTerminalAccessServiceImpl struct {
 	terminalSessionHandler       terminal.TerminalSessionHandler
 	K8sCapacityService           capacity.K8sCapacityService
 	k8sUtil                      *k8s2.K8sServiceImpl
+	asyncRunnable                *async.Runnable
 }
 
 type UserTerminalAccessSessionData struct {
@@ -114,14 +118,15 @@ func NewUserTerminalAccessServiceImpl(logger *zap.SugaredLogger, terminalAccessR
 		terminalSessionHandler:       terminalSessionHandler,
 		K8sCapacityService:           K8sCapacityService,
 		k8sUtil:                      k8sUtil,
+		asyncRunnable:                asyncProvider.NewAsyncRunnable(logger),
 	}
 	podStatusSyncCron.Start()
 	_, err := podStatusSyncCron.AddFunc(fmt.Sprintf("@every %ds", config.TerminalPodStatusSyncTimeInSecs), accessServiceImpl.SyncPodStatus)
 	if err != nil {
 		logger.Errorw("error occurred while starting cron job", "time in secs", config.TerminalPodStatusSyncTimeInSecs)
 		return nil, err
 	}
-	go accessServiceImpl.SyncRunningInstances()
+	accessServiceImpl.asyncRunnable.Execute(func() { accessServiceImpl.SyncRunningInstances() })
 	return accessServiceImpl, err
 }
 func (impl *UserTerminalAccessServiceImpl) ValidateShell(podName, namespace, shellName, containerName string, clusterId int) (bool, string, error) {

pkg/deployment/deployedApp/status/resourceTree/ResourceTreeService.go

@@ -19,6 +19,10 @@ package resourceTree
 import (
 	"context"
 	"fmt"
+	"github.com/devtron-labs/common-lib/async"
+	"strconv"
+	"time"
+
 	"github.com/argoproj/argo-cd/v2/pkg/apiclient/application"
 	"github.com/argoproj/gitops-engine/pkg/health"
 	k8sCommonBean "github.com/devtron-labs/common-lib/utils/k8s/commonBean"
@@ -42,8 +46,6 @@ import (
 	application2 "github.com/devtron-labs/devtron/pkg/k8s/application"
 	util2 "github.com/devtron-labs/devtron/util"
 	"go.uber.org/zap"
-	"strconv"
-	"time"
 )
 
 type Service interface {
@@ -66,6 +68,7 @@ type ServiceImpl struct {
 	k8sApplicationService            application2.K8sApplicationService
 	k8sCommonService                 k8s.K8sCommonService
 	environmentReadService           read2.EnvironmentReadService
+	asyncRunnable                    *async.Runnable
 }
 
 func NewServiceImpl(logger *zap.SugaredLogger,
@@ -78,6 +81,7 @@ func NewServiceImpl(logger *zap.SugaredLogger,
 	k8sApplicationService application2.K8sApplicationService,
 	k8sCommonService k8s.K8sCommonService,
 	environmentReadService read2.EnvironmentReadService,
+	asyncRunnable *async.Runnable,
 ) *ServiceImpl {
 	serviceImpl := &ServiceImpl{
 		logger:                           logger,
@@ -90,6 +94,7 @@ func NewServiceImpl(logger *zap.SugaredLogger,
 		k8sApplicationService:            k8sApplicationService,
 		k8sCommonService:                 k8sCommonService,
 		environmentReadService:           environmentReadService,
+		asyncRunnable:                    asyncRunnable,
 	}
 	return serviceImpl
 }
@@ -164,7 +169,7 @@ func (impl *ServiceImpl) FetchResourceTree(ctx context.Context, appId int, envId
 			}
 		}
 		resourceTree = util2.InterfaceToMapAdapter(resp)
-		go func() {
+		impl.asyncRunnable.Execute(func() {
 			if resp.Status == string(health.HealthStatusHealthy) {
 				err = impl.cdApplicationStatusUpdateHandler.SyncPipelineStatusForResourceTreeCall(cdPipeline)
 				if err != nil {
@@ -176,7 +181,7 @@ func (impl *ServiceImpl) FetchResourceTree(ctx context.Context, appId int, envId
 			if err != nil {
 				impl.logger.Warnw("error in updating app status", "err", err, "appId", cdPipeline.AppId, "envId", cdPipeline.EnvironmentId)
 			}
-		}()
+		})
 		k8sAppDetail := AppView.AppDetailContainer{
 			DeploymentDetailContainer: AppView.DeploymentDetailContainer{
 				ClusterId: cdPipeline.Environment.ClusterId,

pkg/deployment/trigger/devtronApps/HandlerService.go

@@ -19,6 +19,10 @@ package devtronApps
 import (
 	"bufio"
 	"context"
+	"os"
+	"time"
+
+	"github.com/devtron-labs/common-lib/async"
 	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	util5 "github.com/devtron-labs/common-lib/utils/k8s"
 	bean3 "github.com/devtron-labs/devtron/api/bean"
@@ -36,6 +40,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/app"
 	bean4 "github.com/devtron-labs/devtron/pkg/app/bean"
 	"github.com/devtron-labs/devtron/pkg/app/status"
+	"github.com/devtron-labs/devtron/pkg/asyncProvider"
 	"github.com/devtron-labs/devtron/pkg/attributes"
 	"github.com/devtron-labs/devtron/pkg/auth/user"
 	userBean "github.com/devtron-labs/devtron/pkg/auth/user/bean"
@@ -70,8 +75,6 @@ import (
 	util2 "github.com/devtron-labs/devtron/util/event"
 	"github.com/devtron-labs/devtron/util/rbac"
 	"go.uber.org/zap"
-	"os"
-	"time"
 )
 
 /*
@@ -166,6 +169,7 @@ type HandlerServiceImpl struct {
 	ciLogService                        pipeline.CiLogService
 	workflowService                     executor.WorkflowService
 	blobConfigStorageService            pipeline.BlobStorageConfigService
+	asyncRunnable                       *async.Runnable
 }
 
 func NewHandlerServiceImpl(logger *zap.SugaredLogger,
@@ -293,6 +297,7 @@ func NewHandlerServiceImpl(logger *zap.SugaredLogger,
 		ciLogService:             ciLogService,
 		workflowService:          workflowService,
 		blobConfigStorageService: blobConfigStorageService,
+		asyncRunnable:            asyncProvider.NewAsyncRunnable(logger),
 	}
 	config, err := types.GetCdConfig()
 	if err != nil {

pkg/deployment/trigger/devtronApps/deployStageHandlerCode.go

@@ -20,6 +20,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
+	"path"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
 	bean3 "github.com/devtron-labs/devtron/api/bean"
 	"github.com/devtron-labs/devtron/api/bean/gitOps"
 	bean6 "github.com/devtron-labs/devtron/api/helm-app/bean"
@@ -57,12 +64,6 @@ import (
 	"google.golang.org/grpc/codes"
 	status2 "google.golang.org/grpc/status"
 	"helm.sh/helm/v3/pkg/chart"
-	"net/http"
-	"path"
-	"regexp"
-	"strconv"
-	"strings"
-	"time"
 )
 
 func (impl *HandlerServiceImpl) TriggerStageForBulk(triggerRequest bean.TriggerRequest) error {
@@ -741,7 +742,9 @@ func (impl *HandlerServiceImpl) triggerPipeline(overrideRequest *bean3.ValuesOve
 		}
 	}
 
-	go impl.writeCDTriggerEvent(overrideRequest, valuesOverrideResponse.Artifact, valuesOverrideResponse.PipelineOverride.PipelineReleaseCounter, valuesOverrideResponse.PipelineOverride.Id, overrideRequest.WfrId)
+	impl.asyncRunnable.Execute(func() {
+		impl.writeCDTriggerEvent(overrideRequest, valuesOverrideResponse.Artifact, valuesOverrideResponse.PipelineOverride.PipelineReleaseCounter, valuesOverrideResponse.PipelineOverride.Id, overrideRequest.WfrId)
+	})
 
 	_ = impl.markImageScanDeployed(newCtx, overrideRequest.AppId, overrideRequest.EnvId, overrideRequest.ClusterId,
 		valuesOverrideResponse.Artifact.ImageDigest, valuesOverrideResponse.Artifact.ScanEnabled, valuesOverrideResponse.Artifact.Image)

pkg/eventProcessor/in/WorkflowEventProcessorService.go

@@ -21,7 +21,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"slices"
+	"strconv"
+	"sync"
+	"time"
+
 	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"github.com/devtron-labs/common-lib/async"
+	commonConstants "github.com/devtron-labs/common-lib/constants"
 	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/common-lib/pubsub-lib/model"
 	"github.com/devtron-labs/common-lib/utils/registry"
@@ -63,10 +70,6 @@ import (
 	"go.uber.org/zap"
 	"gopkg.in/go-playground/validator.v9"
 	"k8s.io/utils/pointer"
-	"slices"
-	"strconv"
-	"sync"
-	"time"
 )
 
 type WorkflowEventProcessorImpl struct {
@@ -90,6 +93,7 @@ type WorkflowEventProcessorImpl struct {
 	cdPipelineConfigService      pipeline.CdPipelineConfigService
 	userDeploymentRequestService service.UserDeploymentRequestService
 	ucid                         ucid.Service
+	asyncRunnable                *async.Runnable
 
 	devtronAppReleaseContextMap     map[int]bean.DevtronAppReleaseContextType
 	devtronAppReleaseContextMapLock *sync.Mutex
@@ -156,14 +160,15 @@ func NewWorkflowEventProcessorImpl(logger *zap.SugaredLogger,
 		cdWorkflowRepository:            cdWorkflowRepository,
 		deploymentConfigService:         deploymentConfigService,
 		ciHandlerService:                ciHandlerService,
+		asyncRunnable:                   async.NewAsyncRunnable(logger, commonConstants.Orchestrator),
 	}
 	appServiceConfig, err := app.GetAppServiceConfig()
 	if err != nil {
 		return nil, err
 	}
 	impl.appServiceConfig = appServiceConfig
 	// handle incomplete deployment requests after restart
-	go impl.ProcessIncompleteDeploymentReq()
+	impl.asyncRunnable.Execute(func() { impl.ProcessIncompleteDeploymentReq() })
 	return impl, nil
 }