fix: bulk cd deployment trigger failing

api/bean/ValuesOverrideRequest.go

@@ -42,7 +42,7 @@ type ValuesOverrideRequest struct {
 	PipelineId                            int                         `json:"pipelineId" validate:"required"`
 	AppId                                 int                         `json:"appId" validate:"required"`
 	CiArtifactId                          int                         `json:"ciArtifactId" validate:"required"`
-	AdditionalOverride                    json.RawMessage             `json:"additionalOverride"`
+	AdditionalOverride                    json.RawMessage             `json:"additionalOverride,omitempty"`
 	TargetDbVersion                       int                         `json:"targetDbVersion"`
 	ForceTrigger                          bool                        `json:"forceTrigger,notnull"`
 	DeploymentTemplate                    string                      `json:"strategy,omitempty"` // validate:"oneof=BLUE-GREEN ROLLING"`
@@ -54,6 +54,11 @@ type ValuesOverrideRequest struct {
 	DeploymentType                        models.DeploymentType       `json:"-"`
 }
 
+type BulkCdDeployEvent struct {
+	ValuesOverrideRequest *ValuesOverrideRequest `json:"valuesOverrideRequest"`
+	UserId                int32                  `json:"userId"`
+}
+
 type ReleaseStatusUpdateRequest struct {
 	RequestId string             `json:"requestId"`
 	NewStatus models.ChartStatus `json:"newStatus"`

api/restHandler/BulkUpdateRestHandler.go

@@ -332,6 +332,12 @@ func (handler BulkUpdateRestHandlerImpl) BulkDeploy(w http.ResponseWriter, r *ht
 		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
 		return
 	}
+	user, err := handler.userAuthService.GetById(userId)
+	if userId == 0 || err != nil {
+		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
+		return
+	}
+	userEmailId := strings.ToLower(user.EmailId)
 	decoder := json.NewDecoder(r.Body)
 	var request bulkAction.BulkApplicationForEnvironmentPayload
 	err = decoder.Decode(&request)
@@ -346,15 +352,7 @@ func (handler BulkUpdateRestHandlerImpl) BulkDeploy(w http.ResponseWriter, r *ht
 		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
 		return
 	}
-	acdToken, err := handler.argoUserService.GetLatestDevtronArgoCdUserToken()
-	if err != nil {
-		handler.logger.Errorw("error in getting acd token", "err", err)
-		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
-		return
-	}
-	ctx := context.WithValue(r.Context(), "token", acdToken)
-	token := r.Header.Get("token")
-	response, err := handler.bulkUpdateService.BulkDeploy(&request, ctx, w, token, handler.checkAuthForBulkActions)
+	response, err := handler.bulkUpdateService.BulkDeploy(&request, userEmailId, handler.checkAuthBatch)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -489,3 +487,15 @@ func (handler BulkUpdateRestHandlerImpl) HandleCdPipelineBulkAction(w http.Respo
 	}
 	common.WriteJsonResp(w, nil, resp, http.StatusOK)
 }
+
+func (handler BulkUpdateRestHandlerImpl) checkAuthBatch(emailId string, appObject []string, envObject []string) (map[string]bool, map[string]bool) {
+	var appResult map[string]bool
+	var envResult map[string]bool
+	if len(appObject) > 0 {
+		appResult = handler.enforcer.EnforceByEmailInBatch(emailId, casbin.ResourceApplications, casbin.ActionGet, appObject)
+	}
+	if len(envObject) > 0 {
+		envResult = handler.enforcer.EnforceByEmailInBatch(emailId, casbin.ResourceEnvironment, casbin.ActionGet, envObject)
+	}
+	return appResult, envResult
+}

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/coreos/go-oidc v2.2.1+incompatible
 	github.com/davecgh/go-spew v1.1.1
 	github.com/devtron-labs/authenticator v0.4.31-0.20221213131053-6e4668309f53
-	github.com/devtron-labs/common-lib v0.0.0-20230228120609-58fa7d3de8e2
+	github.com/devtron-labs/common-lib v0.0.0-20230407072229-d4f665f5ca12
 	github.com/evanphx/json-patch v5.6.0+incompatible
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32
 	github.com/go-pg/pg v6.15.1+incompatible

go.sum

@@ -282,22 +282,8 @@ github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4 h1:YcpmyvADG
 github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4/go.mod h1:zAg7JM8CkOJ43xKXIj7eRO9kmWm/TW578qo+oDO6tuM=
 github.com/devtron-labs/authenticator v0.4.31-0.20221213131053-6e4668309f53 h1:oHDpsCsuYiL+u5TEhilKNLGhH9lwRNkIqrzhHQA6d3c=
 github.com/devtron-labs/authenticator v0.4.31-0.20221213131053-6e4668309f53/go.mod h1:ozNfT8WcruiSgnUbyp48WVfc41++W6xYXhKFp67lNTU=
-github.com/devtron-labs/common-lib v0.0.0-20230109070754-ff4dca200a2c h1:jz7yPUlIJXFg9AvJh2fb0QW7JT6+RKj8LOl1mWM/HQA=
-github.com/devtron-labs/common-lib v0.0.0-20230109070754-ff4dca200a2c/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230202071119-47116fd18242 h1:FfS2R86Ft/WFShnwGu4z14iOKZAPhfDV7tvvlaJfRmI=
-github.com/devtron-labs/common-lib v0.0.0-20230202071119-47116fd18242/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230202123846-6a2fbc3adf9a h1:W7iFtOdOj/pYLBLkVxkpXIGX8La5nCq3tZey1wtSeCo=
-github.com/devtron-labs/common-lib v0.0.0-20230202123846-6a2fbc3adf9a/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230202130856-d4ef0e6ba405 h1:PHtTohiRePmGWkNCykvLY6Mpebke4GlW4fvO9zHzn5E=
-github.com/devtron-labs/common-lib v0.0.0-20230202130856-d4ef0e6ba405/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230202133846-0be4c43dd7f2 h1:O0NgzYI+dF2OxncbzfcsscIPV6Kf1G6O3SXec7yhs7E=
-github.com/devtron-labs/common-lib v0.0.0-20230202133846-0be4c43dd7f2/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230220060258-e0dba0258e97 h1:3gzBIn0k/4/Cal0px0HuzOcbDP/12C8U+nUN0GzolIA=
-github.com/devtron-labs/common-lib v0.0.0-20230220060258-e0dba0258e97/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230220122715-29f6f63d39e9 h1:JKaG846oVWXdx+1zqb777sYEhvmYOEJOI+Wroc4AQaI=
-github.com/devtron-labs/common-lib v0.0.0-20230220122715-29f6f63d39e9/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
-github.com/devtron-labs/common-lib v0.0.0-20230228120609-58fa7d3de8e2 h1:Z4OIQU6mTBZREYBN0+GEl7E2Y4ffWCWPbu9OSowbZlE=
-github.com/devtron-labs/common-lib v0.0.0-20230228120609-58fa7d3de8e2/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
+github.com/devtron-labs/common-lib v0.0.0-20230407072229-d4f665f5ca12 h1:tcwNgAWTzalItF5XT0WxicRvY4wzz/o0HYLMVLxsxjg=
+github.com/devtron-labs/common-lib v0.0.0-20230407072229-d4f665f5ca12/go.mod h1:R24nOqgk4buk9zv+BXzORfObZsOe3NE9P55KrZXGX9k=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=

pkg/bulkAction/BulkUpdateService.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/devtron/api/bean"
 	client "github.com/devtron-labs/devtron/api/helm-app"
 	openapi "github.com/devtron-labs/devtron/api/helm-app/openapiClient"
@@ -46,7 +47,7 @@ type BulkUpdateService interface {
 
 	BulkHibernate(request *BulkApplicationForEnvironmentPayload, ctx context.Context, w http.ResponseWriter, token string, checkAuthForBulkActions func(token string, appObject string, envObject string) bool) (*BulkApplicationForEnvironmentResponse, error)
 	BulkUnHibernate(request *BulkApplicationForEnvironmentPayload, ctx context.Context, w http.ResponseWriter, token string, checkAuthForBulkActions func(token string, appObject string, envObject string) bool) (*BulkApplicationForEnvironmentResponse, error)
-	BulkDeploy(request *BulkApplicationForEnvironmentPayload, ctx context.Context, w http.ResponseWriter, token string, checkAuthForBulkActions func(token string, appObject string, envObject string) bool) (*BulkApplicationForEnvironmentResponse, error)
+	BulkDeploy(request *BulkApplicationForEnvironmentPayload, emailId string, checkAuthBatch func(emailId string, appObject []string, envObject []string) (map[string]bool, map[string]bool)) (*BulkApplicationForEnvironmentResponse, error)
 	BulkBuildTrigger(request *BulkApplicationForEnvironmentPayload, ctx context.Context, w http.ResponseWriter, token string, checkAuthForBulkActions func(token string, appObject string, envObject string) bool) (*BulkApplicationForEnvironmentResponse, error)
 
 	GetBulkActionImpactedPipelinesAndWfs(dto *CdBulkActionRequestDto) ([]*pipelineConfig.Pipeline, []int, []int, error)
@@ -84,6 +85,7 @@ type BulkUpdateServiceImpl struct {
 	ciPipelineRepository             pipelineConfig.CiPipelineRepository
 	appWorkflowRepository            appWorkflow.AppWorkflowRepository
 	appWorkflowService               appWorkflow2.AppWorkflowService
+	pubsubClient                     *pubsub.PubSubClientServiceImpl
 }
 
 func NewBulkUpdateServiceImpl(bulkUpdateRepository bulkUpdate.BulkUpdateRepository,
@@ -111,8 +113,9 @@ func NewBulkUpdateServiceImpl(bulkUpdateRepository bulkUpdate.BulkUpdateReposito
 	enforcerUtilHelm rbac.EnforcerUtilHelm, ciHandler pipeline.CiHandler,
 	ciPipelineRepository pipelineConfig.CiPipelineRepository,
 	appWorkflowRepository appWorkflow.AppWorkflowRepository,
-	appWorkflowService appWorkflow2.AppWorkflowService) *BulkUpdateServiceImpl {
-	return &BulkUpdateServiceImpl{
+	appWorkflowService appWorkflow2.AppWorkflowService,
+	pubsubClient *pubsub.PubSubClientServiceImpl) (*BulkUpdateServiceImpl, error) {
+	impl := &BulkUpdateServiceImpl{
 		bulkUpdateRepository:             bulkUpdateRepository,
 		chartRepository:                  chartRepository,
 		logger:                           logger,
@@ -143,7 +146,11 @@ func NewBulkUpdateServiceImpl(bulkUpdateRepository bulkUpdate.BulkUpdateReposito
 		ciPipelineRepository:             ciPipelineRepository,
 		appWorkflowRepository:            appWorkflowRepository,
 		appWorkflowService:               appWorkflowService,
+		pubsubClient:                     pubsubClient,
 	}
+
+	err := impl.SubscribeToCdBulkTriggerTopic()
+	return impl, err
 }
 
 func (impl BulkUpdateServiceImpl) FindBulkUpdateReadme(operation string) (*BulkUpdateSeeExampleResponse, error) {
@@ -1138,7 +1145,7 @@ func (impl BulkUpdateServiceImpl) BulkUnHibernate(request *BulkApplicationForEnv
 	bulkOperationResponse.Response = response
 	return bulkOperationResponse, nil
 }
-func (impl BulkUpdateServiceImpl) BulkDeploy(request *BulkApplicationForEnvironmentPayload, ctx context.Context, w http.ResponseWriter, token string, checkAuthForBulkActions func(token string, appObject string, envObject string) bool) (*BulkApplicationForEnvironmentResponse, error) {
+func (impl BulkUpdateServiceImpl) BulkDeploy(request *BulkApplicationForEnvironmentPayload, emailId string, checkAuthBatch func(emailId string, appObject []string, envObject []string) (map[string]bool, map[string]bool)) (*BulkApplicationForEnvironmentResponse, error) {
 	var pipelines []*pipelineConfig.Pipeline
 	var err error
 	if len(request.AppIdIncludes) > 0 {
@@ -1152,6 +1159,26 @@ func (impl BulkUpdateServiceImpl) BulkDeploy(request *BulkApplicationForEnvironm
 		impl.logger.Errorw("error in fetching pipelines", "envId", request.EnvId, "err", err)
 		return nil, err
 	}
+
+	pipelineIds := make([]int, 0)
+	for _, pipeline := range pipelines {
+		pipelineIds = append(pipelineIds, pipeline.Id)
+	}
+	if len(pipelineIds) == 0 {
+		return nil, fmt.Errorf("no pipeline found for this environment")
+	}
+	//authorization block starts here
+	var appObjectArr []string
+	var envObjectArr []string
+	objects := impl.enforcerUtil.GetAppAndEnvObjectByPipelineIds(pipelineIds)
+	pipelineIds = []int{}
+	for _, object := range objects {
+		appObjectArr = append(appObjectArr, object[0])
+		envObjectArr = append(envObjectArr, object[1])
+	}
+	appResults, envResults := checkAuthBatch(emailId, appObjectArr, envObjectArr)
+	//authorization block ends here
+
 	response := make(map[string]map[string]bool)
 	for _, pipeline := range pipelines {
 		appKey := fmt.Sprintf("%d_%s", pipeline.AppId, pipeline.App.AppName)
@@ -1162,11 +1189,10 @@ func (impl BulkUpdateServiceImpl) BulkDeploy(request *BulkApplicationForEnvironm
 			pResponse[pipelineKey] = false
 			response[appKey] = pResponse
 		}
-		appObject := impl.enforcerUtil.GetAppRBACNameByAppId(pipeline.AppId)
-		envObject := impl.enforcerUtil.GetEnvRBACNameByAppId(pipeline.AppId, pipeline.EnvironmentId)
-		isValidAuth := checkAuthForBulkActions(token, appObject, envObject)
-		if !isValidAuth {
-			//skip hibernate for the app if user does not have access on that
+		appObject := objects[pipeline.Id][0]
+		envObject := objects[pipeline.Id][1]
+		if !(appResults[appObject] && envResults[envObject]) {
+			//if user unauthorized, skip items
 			pipelineResponse := response[appKey]
 			pipelineResponse[pipelineKey] = false
 			response[appKey] = pipelineResponse
@@ -1198,14 +1224,36 @@ func (impl BulkUpdateServiceImpl) BulkDeploy(request *BulkApplicationForEnvironm
 			UserId:         request.UserId,
 			CdWorkflowType: bean.CD_WORKFLOW_TYPE_DEPLOY,
 		}
-		_, err = impl.workflowDagExecutor.ManualCdTrigger(overrideRequest, ctx)
+		event := &bean.BulkCdDeployEvent{
+			ValuesOverrideRequest: overrideRequest,
+			UserId:                overrideRequest.UserId,
+		}
+
+		payload, err := json.Marshal(event)
 		if err != nil {
-			impl.logger.Errorw("request err, OverrideConfig", "err", err, "payload", overrideRequest)
+			impl.logger.Errorw("failed to marshal cd bulk deploy event request",
+				"request", event,
+				"err", err)
+
 			pipelineResponse := response[appKey]
 			pipelineResponse[pipelineKey] = false
 			response[appKey] = pipelineResponse
-			//return nil, err
+			continue
+		}
+
+		err = impl.pubsubClient.Publish(pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC, string(payload))
+		if err != nil {
+			impl.logger.Errorw("failed to publish trigger request event",
+				"topic", pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC,
+				"request", overrideRequest,
+				"err", err)
+
+			pipelineResponse := response[appKey]
+			pipelineResponse[pipelineKey] = false
+			response[appKey] = pipelineResponse
+			continue
 		}
+
 		pipelineResponse := response[appKey]
 		pipelineResponse[pipelineKey] = success
 		response[appKey] = pipelineResponse
@@ -1216,6 +1264,43 @@ func (impl BulkUpdateServiceImpl) BulkDeploy(request *BulkApplicationForEnvironm
 	return bulkOperationResponse, nil
 }
 
+func (impl BulkUpdateServiceImpl) SubscribeToCdBulkTriggerTopic() error {
+
+	callback := func(msg *pubsub.PubSubMsg) {
+		impl.logger.Infow("Event received",
+			"topic", pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC,
+			"msg", msg.Data)
+
+		event := &bean.BulkCdDeployEvent{}
+		err := json.Unmarshal([]byte(msg.Data), event)
+		if err != nil {
+			impl.logger.Errorw("Error unmarshalling received event",
+				"topic", pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC,
+				"msg", msg.Data,
+				"err", err)
+			return
+		}
+		event.ValuesOverrideRequest.UserId = event.UserId
+
+		// trigger
+		_, err = impl.workflowDagExecutor.ManualCdTrigger(event.ValuesOverrideRequest, context.Background())
+		if err != nil {
+			impl.logger.Errorw("Error triggering CD",
+				"topic", pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC,
+				"msg", msg.Data,
+				"err", err)
+		}
+	}
+	err := impl.pubsubClient.Subscribe(pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC, callback)
+	if err != nil {
+		impl.logger.Error("failed to subscribe to NATS topic",
+			"topic", pubsub.CD_BULK_DEPLOY_TRIGGER_TOPIC,
+			"err", err)
+		return err
+	}
+	return nil
+}
+
 func (impl BulkUpdateServiceImpl) BulkBuildTrigger(request *BulkApplicationForEnvironmentPayload, ctx context.Context, w http.ResponseWriter, token string, checkAuthForBulkActions func(token string, appObject string, envObject string) bool) (*BulkApplicationForEnvironmentResponse, error) {
 	var pipelines []*pipelineConfig.Pipeline
 	var err error