feat: upload and download logs/artifact from blob storage configured in external cluster

pkg/pipeline/CdHandler.go

@@ -20,6 +20,7 @@ package pipeline
 import (
 	"bufio"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	application2 "github.com/argoproj/argo-cd/v2/pkg/apiclient/application"
@@ -52,6 +53,8 @@ import (
 	"github.com/go-pg/pg"
 	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
+	errors2 "k8s.io/apimachinery/pkg/api/errors"
+	v12 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/rest"
 	"os"
 	"path/filepath"
@@ -119,9 +122,10 @@ type CdHandlerImpl struct {
 	k8sUtil                                *k8s.K8sUtil
 	workflowService                        WorkflowService
 	config                                 *CdConfig
+	clusterRepository                      repository2.ClusterRepository
 }
 
-func NewCdHandlerImpl(Logger *zap.SugaredLogger, userService user.UserService, cdWorkflowRepository pipelineConfig.CdWorkflowRepository, ciLogService CiLogService, ciArtifactRepository repository.CiArtifactRepository, ciPipelineMaterialRepository pipelineConfig.CiPipelineMaterialRepository, pipelineRepository pipelineConfig.PipelineRepository, envRepository repository2.EnvironmentRepository, ciWorkflowRepository pipelineConfig.CiWorkflowRepository, helmAppService client.HelmAppService, pipelineOverrideRepository chartConfig.PipelineOverrideRepository, workflowDagExecutor WorkflowDagExecutor, appListingService app.AppListingService, appListingRepository repository.AppListingRepository, pipelineStatusTimelineRepository pipelineConfig.PipelineStatusTimelineRepository, application application.ServiceClient, argoUserService argo.ArgoUserService, deploymentEventHandler app.DeploymentEventHandler, eventClient client2.EventClient, pipelineStatusTimelineResourcesService status.PipelineStatusTimelineResourcesService, pipelineStatusSyncDetailService status.PipelineStatusSyncDetailService, pipelineStatusTimelineService status.PipelineStatusTimelineService, appService app.AppService, appStatusService app_status.AppStatusService, enforcerUtil rbac.EnforcerUtil, installedAppRepository repository3.InstalledAppRepository, installedAppVersionHistoryRepository repository3.InstalledAppVersionHistoryRepository, appRepository app2.AppRepository, resourceGroupService resourceGroup2.ResourceGroupService, imageTaggingService ImageTaggingService, k8sUtil *k8s.K8sUtil, workflowService WorkflowService) *CdHandlerImpl {
+func NewCdHandlerImpl(Logger *zap.SugaredLogger, userService user.UserService, cdWorkflowRepository pipelineConfig.CdWorkflowRepository, ciLogService CiLogService, ciArtifactRepository repository.CiArtifactRepository, ciPipelineMaterialRepository pipelineConfig.CiPipelineMaterialRepository, pipelineRepository pipelineConfig.PipelineRepository, envRepository repository2.EnvironmentRepository, ciWorkflowRepository pipelineConfig.CiWorkflowRepository, helmAppService client.HelmAppService, pipelineOverrideRepository chartConfig.PipelineOverrideRepository, workflowDagExecutor WorkflowDagExecutor, appListingService app.AppListingService, appListingRepository repository.AppListingRepository, pipelineStatusTimelineRepository pipelineConfig.PipelineStatusTimelineRepository, application application.ServiceClient, argoUserService argo.ArgoUserService, deploymentEventHandler app.DeploymentEventHandler, eventClient client2.EventClient, pipelineStatusTimelineResourcesService status.PipelineStatusTimelineResourcesService, pipelineStatusSyncDetailService status.PipelineStatusSyncDetailService, pipelineStatusTimelineService status.PipelineStatusTimelineService, appService app.AppService, appStatusService app_status.AppStatusService, enforcerUtil rbac.EnforcerUtil, installedAppRepository repository3.InstalledAppRepository, installedAppVersionHistoryRepository repository3.InstalledAppVersionHistoryRepository, appRepository app2.AppRepository, resourceGroupService resourceGroup2.ResourceGroupService, imageTaggingService ImageTaggingService, k8sUtil *k8s.K8sUtil, workflowService WorkflowService, clusterRepository repository2.ClusterRepository) *CdHandlerImpl {
 	cdh := &CdHandlerImpl{
 		Logger:                                 Logger,
 		userService:                            userService,
@@ -155,6 +159,7 @@ func NewCdHandlerImpl(Logger *zap.SugaredLogger, userService user.UserService, c
 		imageTaggingService:                    imageTaggingService,
 		k8sUtil:                                k8sUtil,
 		workflowService:                        workflowService,
+		clusterRepository:                      clusterRepository,
 	}
 	config, err := GetCdConfig()
 	if err != nil {
@@ -170,6 +175,29 @@ type ArgoPipelineStatusSyncEvent struct {
 	IsAppStoreApplication bool  `json:"isAppStoreApplication"`
 }
 
+type CmBlobStorageConfig struct {
+	//AWS credentials
+	CloudProvider      blob_storage.BlobStorageType `json:"BLOB_STORAGE_PROVIDER"`
+	S3AccessKey        string                       `json:"BLOB_STORAGE_S3_ACCESS_KEY"`
+	S3Endpoint         string                       `json:"BLOB_STORAGE_S3_ENDPOINT"`
+	S3EndpointInsecure bool                         `json:"BLOB_STORAGE_S3_ENDPOINT_INSECURE"`
+	S3BucketVersioned  bool                         `json:"BLOB_STORAGE_S3_BUCKET_VERSIONED"`
+
+	//Azure credentials
+	AzureAccountName               string `json:"AZURE_ACCOUNT_NAME"`
+	AzureGatewayUrl                string `json:"AZURE_GATEWAY_URL"`
+	AzureGatewayConnectionInsecure bool   `json:"AZURE_GATEWAY_CONNECTION_INSECURE"`
+}
+
+type SecretBlobStorageConfig struct {
+	//aws
+	S3SecretKey string `json:"BLOB_STORAGE_S3_SECRET_KEY"`
+	//gcp
+	GcpBlobStorageCredentialJson string `json:"BLOB_STORAGE_GCP_CREDENTIALS_JSON"`
+	//azure
+	AzureAccountKey string `json:"AZURE_ACCOUNT_KEY"`
+}
+
 const NotTriggered string = "Not Triggered"
 const NotDeployed = "Not Deployed"
 const WorklowTypeDeploy = "DEPLOY"
@@ -913,7 +941,7 @@ func (impl *CdHandlerImpl) getWorkflowLogs(pipelineId int, cdWorkflow *pipelineC
 			return nil, nil, errors.New("logs-not-stored-in-repository")
 		} else if string(v1alpha1.NodeSucceeded) == cdWorkflow.Status || string(v1alpha1.NodeError) == cdWorkflow.Status || string(v1alpha1.NodeFailed) == cdWorkflow.Status || cdWorkflow.Status == WorkflowCancel {
 			impl.Logger.Debugw("pod is not live ", "err", err)
-			return impl.getLogsFromRepository(pipelineId, cdWorkflow)
+			return impl.getLogsFromRepository(pipelineId, cdWorkflow, clusterConfig, runStageInEnv)
 		}
 		impl.Logger.Errorw("err on fetch workflow logs", "err", err)
 		return nil, nil, err
@@ -922,7 +950,7 @@ func (impl *CdHandlerImpl) getWorkflowLogs(pipelineId int, cdWorkflow *pipelineC
 	return logReader, cleanUp, err
 }
 
-func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pipelineConfig.CdWorkflowRunner) (*bufio.Reader, func() error, error) {
+func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pipelineConfig.CdWorkflowRunner, clusterConfig *k8s.ClusterConfig, isExt bool) (*bufio.Reader, func() error, error) {
 	impl.Logger.Debug("getting historic logs")
 
 	cdConfig, err := impl.cdWorkflowRepository.FindConfigByPipelineId(pipelineId)
@@ -964,6 +992,29 @@ func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pip
 			CredentialFileJsonData: impl.config.BlobStorageGcpCredentialJson,
 		},
 	}
+
+	useExternalBlobStorage := isExt && !impl.config.UseBlobStorageConfigInCdWorkflow
+	if useExternalBlobStorage {
+		//fetch extClusterBlob cm and cs from k8s client, if they are present then read creds
+		//from them else return.
+		cmConfig, secretConfig, err := impl.FetchCmAndSecretBlobConfigFromExternalCluster(clusterConfig, cdWorkflow.Namespace)
+		if err != nil {
+			impl.Logger.Errorw("error in fetching config map and secret from external cluster", "err", err, "clusterConfig", clusterConfig)
+			return nil, nil, err
+		}
+		cdLogRequest.CloudProvider = cmConfig.CloudProvider
+		cdLogRequest.AzureBlobConfig.AccountName = cmConfig.AzureAccountName
+		cdLogRequest.AzureBlobConfig.AccountKey = secretConfig.AzureAccountKey
+
+		cdLogRequest.GcpBlobBaseConfig.CredentialFileJsonData = secretConfig.GcpBlobStorageCredentialJson
+
+		cdLogRequest.AwsS3BaseConfig.AccessKey = cmConfig.S3AccessKey
+		cdLogRequest.AwsS3BaseConfig.EndpointUrl = cmConfig.S3Endpoint
+		cdLogRequest.AwsS3BaseConfig.IsInSecure = cmConfig.S3EndpointInsecure
+		cdLogRequest.AwsS3BaseConfig.Passkey = secretConfig.S3SecretKey
+
+	}
+
 	impl.Logger.Infow("s3 log req ", "req", cdLogRequest)
 	oldLogsStream, cleanUp, err := impl.ciLogService.FetchLogs(impl.config.BaseLogLocationPath, cdLogRequest)
 	if err != nil {
@@ -974,6 +1025,59 @@ func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pip
 	return logReader, cleanUp, err
 }
 
+func (impl *CdHandlerImpl) FetchCmAndSecretBlobConfigFromExternalCluster(clusterConfig *k8s.ClusterConfig, namespace string) (CmBlobStorageConfig, SecretBlobStorageConfig, error) {
+	var cmConfig CmBlobStorageConfig
+	var secretConfig SecretBlobStorageConfig
+	_, _, kubeClient, err := impl.k8sUtil.GetK8sConfigAndClients(clusterConfig)
+	if err != nil {
+		impl.Logger.Errorw("FetchCmAndSecretBlobConfigFromExternalCluster, error in getting kubeClient by cluster config", "err", err)
+		return cmConfig, secretConfig, err
+	}
+	cm, err := kubeClient.CoreV1().ConfigMaps(namespace).Get(context.Background(), impl.config.ExtBlobStorageCmName, v12.GetOptions{})
+	if err != nil && !errors2.IsNotFound(err) {
+		impl.Logger.Errorw("error in getting config map in external cluster", "err", err, "blobStorageCmName", impl.config.ExtBlobStorageCmName, "clusterName", clusterConfig.ClusterName)
+		return cmConfig, secretConfig, err
+	}
+	if errors2.IsNotFound(err) {
+		cmNotFoundErr := fmt.Sprintf("config map: %s not found in namespace: %s", impl.config.ExtBlobStorageCmName, namespace)
+		return cmConfig, secretConfig, errors.New(cmNotFoundErr)
+	}
+	secret, err := kubeClient.CoreV1().Secrets(namespace).Get(context.Background(), impl.config.ExtBlobStorageSecretName, v12.GetOptions{})
+	if err != nil && !errors2.IsNotFound(err) {
+		impl.Logger.Errorw("error in getting secret in external cluster", "err", err, "blobStorageSecretName", impl.config.ExtBlobStorageSecretName, "clusterName", clusterConfig.ClusterName)
+		return cmConfig, secretConfig, err
+	}
+	if errors2.IsNotFound(err) {
+		secretNotFoundErr := fmt.Sprintf("secret: %s not found in namespace: %s", impl.config.ExtBlobStorageSecretName, namespace)
+		return cmConfig, secretConfig, errors.New(secretNotFoundErr)
+	}
+	if cm.Data != nil && secret != nil {
+		cmDataJson, err := json.Marshal(cm.Data)
+		if err != nil {
+			fmt.Println("error marshalling external blob storage cm data to json:", err)
+			return cmConfig, secretConfig, err
+		}
+		err = json.Unmarshal(cmDataJson, &cmConfig)
+		if err != nil {
+			fmt.Println("error unmarshalling external blob storage cm json to struct:", err)
+			return cmConfig, secretConfig, err
+		}
+
+		secretDataJson, err := json.Marshal(secret.Data)
+		if err != nil {
+			fmt.Println("error marshalling external blob storage secret data to json:", err)
+			return cmConfig, secretConfig, err
+		}
+
+		err = json.Unmarshal(secretDataJson, &secretConfig)
+		if err != nil {
+			fmt.Println("error unmarshalling external blob storage secret json to struct:", err)
+			return cmConfig, secretConfig, err
+		}
+	}
+	return cmConfig, secretConfig, nil
+}
+
 func (impl *CdHandlerImpl) FetchCdWorkflowDetails(appId int, environmentId int, pipelineId int, buildId int) (WorkflowResponse, error) {
 	workflowR, err := impl.cdWorkflowRepository.FindWorkflowRunnerById(buildId)
 	if err != nil && err != pg.ErrNoRows {
@@ -1065,7 +1169,27 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 		impl.Logger.Errorw("unable to fetch ciWorkflow", "err", err)
 		return nil, err
 	}
-
+	var clusterBean cluster.ClusterBean
+	if wfr.CdWorkflow.Pipeline.Environment.Id != 0 {
+		clusterFromDb, err := impl.clusterRepository.FindById(wfr.CdWorkflow.Pipeline.Environment.ClusterId)
+		if err != nil {
+			impl.Logger.Errorw("unable to fetch cluster by cluster Id", "clusterId", wfr.CdWorkflow.Pipeline.Environment.ClusterId, "err", err)
+			return nil, err
+		}
+		clusterBean = cluster.GetClusterBean(*clusterFromDb)
+	}
+	clusterConfig, err := clusterBean.GetClusterConfig()
+	if err != nil {
+		impl.Logger.Errorw("error in getting cluster config", "err", err, "clusterId", clusterBean.Id)
+		return nil, err
+	}
+	var isExtCluster bool
+	if wfr.WorkflowType == PRE {
+		isExtCluster = wfr.CdWorkflow.Pipeline.RunPreStageInEnv
+	} else if wfr.WorkflowType == POST {
+		isExtCluster = wfr.CdWorkflow.Pipeline.RunPostStageInEnv
+	}
+	useExternalBlobStorage := isExtCluster && !impl.config.UseBlobStorageConfigInCdWorkflow
 	if !wfr.BlobStorageEnabled {
 		return nil, errors.New("logs-not-stored-in-repository")
 	}
@@ -1115,6 +1239,26 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 		AwsS3BaseConfig:     awsS3BaseConfig,
 		GcpBlobBaseConfig:   gcpBlobBaseConfig,
 	}
+	if useExternalBlobStorage {
+		//fetch extClusterBlob cm and cs from k8s client, if they are present then read creds
+		//from them else return.
+		cmConfig, secretConfig, err := impl.FetchCmAndSecretBlobConfigFromExternalCluster(clusterConfig, wfr.Namespace)
+		if err != nil {
+			impl.Logger.Errorw("error in fetching config map and secret from external cluster", "err", err, "clusterConfig", clusterConfig)
+			return nil, err
+		}
+		request.StorageType = cmConfig.CloudProvider
+
+		request.AwsS3BaseConfig.AccessKey = cmConfig.S3AccessKey
+		request.AwsS3BaseConfig.EndpointUrl = cmConfig.S3Endpoint
+		request.AwsS3BaseConfig.IsInSecure = cmConfig.S3EndpointInsecure
+		request.AwsS3BaseConfig.Passkey = secretConfig.S3SecretKey
+
+		request.AzureBlobBaseConfig.AccountName = cmConfig.AzureAccountName
+		request.AzureBlobBaseConfig.AccountKey = secretConfig.AzureAccountKey
+
+		request.GcpBlobBaseConfig.CredentialFileJsonData = secretConfig.GcpBlobStorageCredentialJson
+	}
 	_, numBytes, err := blobStorageService.Get(request)
 	if err != nil {
 		impl.Logger.Errorw("error occurred while downloading file", "request", request, "error", err)

pkg/pipeline/CiCdConfig.go

@@ -116,6 +116,8 @@ type CiCdConfig struct {
 	BaseLogLocationPath            string                       `env:"BASE_LOG_LOCATION_PATH" envDefault:"/home/devtron/"`
 	InAppLoggingEnabled            bool                         `env:"IN_APP_LOGGING_ENABLED" envDefault:"false"`
 	BuildxProvenanceMode           string                       `env:"BUILDX_PROVENANCE_MODE" envDefault:""` //provenance is set to false if this flag is not set
+	ExtBlobStorageCmName           string                       `env:"EXTERNAL_BLOB_STORAGE_CM_NAME" envDefault:"blob-storage-cm"`
+	ExtBlobStorageSecretName       string                       `env:"EXTERNAL_BLOB_STORAGE_SECRET_NAME" envDefault:"blob-storage-secret"`
 }
 
 type CiConfig struct {

pkg/pipeline/CiService.go

@@ -593,9 +593,12 @@ func (impl *CiServiceImpl) buildWfRequestForCiPipeline(pipeline *pipelineConfig.
 		ImageRetryCount:            impl.config.ImageRetryCount,
 		ImageRetryInterval:         impl.config.ImageRetryInterval,
 		WorkflowExecutor:           impl.config.GetWorkflowExecutorType(),
+		ExtBlobStorageSecretName:   impl.config.ExtBlobStorageSecretName,
+		ExtBlobStorageCmName:       impl.config.ExtBlobStorageCmName,
 		Type:                       bean2.CI_WORKFLOW_PIPELINE_TYPE,
 		CiArtifactLastFetch:        trigger.CiArtifactLastFetch,
 	}
+
 	if dockerRegistry != nil {
 
 		workflowRequest.DockerRegistryId = dockerRegistry.Id

pkg/pipeline/WorkflowDagExecutor.go

@@ -995,11 +995,13 @@ func (impl *WorkflowDagExecutorImpl) buildWFRequest(runner *pipelineConfig.CdWor
 			DataSource:   artifact.DataSource,
 			WorkflowId:   artifact.WorkflowId,
 		},
-		OrchestratorHost:  impl.config.OrchestratorHost,
-		OrchestratorToken: impl.config.OrchestratorToken,
-		CloudProvider:     impl.config.CloudProvider,
-		WorkflowExecutor:  workflowExecutor,
-		RefPlugins:        refPluginsData,
+		OrchestratorHost:         impl.config.OrchestratorHost,
+		OrchestratorToken:        impl.config.OrchestratorToken,
+		CloudProvider:            impl.config.CloudProvider,
+		WorkflowExecutor:         workflowExecutor,
+		RefPlugins:               refPluginsData,
+		ExtBlobStorageSecretName: impl.config.ExtBlobStorageSecretName,
+		ExtBlobStorageCmName:     impl.config.ExtBlobStorageCmName,
 	}
 
 	extraEnvVariables := make(map[string]string)

pkg/pipeline/WorkflowService.go

@@ -116,6 +116,7 @@ func (impl *WorkflowServiceImpl) SubmitWorkflow(workflowRequest *WorkflowRequest
 }
 
 func (impl *WorkflowServiceImpl) createWorkflowTemplate(workflowRequest *WorkflowRequest) (bean3.WorkflowTemplate, error) {
+	workflowRequest.UseExternalClusterBlob = !workflowRequest.CheckBlobStorageConfig(impl.ciCdConfig) && workflowRequest.IsExtRun
 	workflowJson, err := workflowRequest.GetWorkflowJson(impl.ciCdConfig)
 	if err != nil {
 		impl.Logger.Errorw("error occurred while getting workflow json", "err", err)
@@ -243,6 +244,24 @@ func (impl *WorkflowServiceImpl) addExistingCmCsInWorkflow(workflowRequest *Work
 			workflowSecrets = append(workflowSecrets, *secret)
 		}
 	}
+	//internally inducing BlobStorageCmName and BlobStorageSecretName for getting logs, caches and artifacts from
+	//in-cluster configured blob storage, if USE_BLOB_STORAGE_CONFIG_IN_CD_WORKFLOW = false and isExt = true
+	if workflowRequest.UseExternalClusterBlob {
+		blobDetailsConfigMap := bean.ConfigSecretMap{
+			Name:     impl.ciCdConfig.ExtBlobStorageCmName,
+			Type:     "environment",
+			External: true,
+		}
+		workflowConfigMaps = append(workflowConfigMaps, blobDetailsConfigMap)
+
+		blobDetailsSecret := bean.ConfigSecretMap{
+			Name:     impl.ciCdConfig.ExtBlobStorageSecretName,
+			Type:     "environment",
+			External: true,
+		}
+		workflowSecrets = append(workflowSecrets, blobDetailsSecret)
+	}
+
 	return workflowConfigMaps, workflowSecrets, nil
 }
 

pkg/pipeline/WorkflowUtils.go

@@ -366,6 +366,9 @@ type WorkflowRequest struct {
 	WorkflowExecutor         pipelineConfig.WorkflowExecutorType `json:"workflowExecutor"`
 	PrePostDeploySteps       []*bean.StepObject                  `json:"prePostDeploySteps"`
 	CiArtifactLastFetch      time.Time                           `json:"ciArtifactLastFetch"`
+	ExtBlobStorageCmName     string                              `json:"extBlobStorageCmName"`
+	ExtBlobStorageSecretName string                              `json:"extBlobStorageSecretName"`
+	UseExternalClusterBlob   bool                                `json:"useExternalClusterBlob"`
 	Type                     bean.WorkflowPipelineType
 	Pipeline                 *pipelineConfig.Pipeline
 	Env                      *repository2.Environment
@@ -471,10 +474,6 @@ func (workflowRequest *WorkflowRequest) getContainerEnvVariables(config *CiCdCon
 		workflowRequest.Type == bean.JOB_WORKFLOW_PIPELINE_TYPE {
 		containerEnvVariables = []v12.EnvVar{{Name: "IMAGE_SCANNER_ENDPOINT", Value: config.ImageScannerEndpoint}}
 	}
-	if config.CloudProvider == BLOB_STORAGE_S3 && config.BlobStorageS3AccessKey != "" {
-		miniCred := []v12.EnvVar{{Name: "AWS_ACCESS_KEY_ID", Value: config.BlobStorageS3AccessKey}, {Name: "AWS_SECRET_ACCESS_KEY", Value: config.BlobStorageS3SecretKey}}
-		containerEnvVariables = append(containerEnvVariables, miniCred...)
-	}
 	eventEnv := v12.EnvVar{Name: "CI_CD_EVENT", Value: string(workflowJson)}
 	inAppLoggingEnv := v12.EnvVar{Name: "IN_APP_LOGGING", Value: strconv.FormatBool(workflowRequest.InAppLoggingEnabled)}
 	containerEnvVariables = append(containerEnvVariables, eventEnv, inAppLoggingEnv)