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

Wire.go

@@ -571,6 +571,9 @@ func InitializeApp() (*App, error) {
 		pipeline.NewCdHandlerImpl,
 		wire.Bind(new(pipeline.CdHandler), new(*pipeline.CdHandlerImpl)),
 
+		pipeline.NewBlobStorageConfigServiceImpl,
+		wire.Bind(new(pipeline.BlobStorageConfigService), new(*pipeline.BlobStorageConfigServiceImpl)),
+
 		pipeline.NewWorkflowDagExecutorImpl,
 		wire.Bind(new(pipeline.WorkflowDagExecutor), new(*pipeline.WorkflowDagExecutorImpl)),
 		appClone.NewAppCloneServiceImpl,

internal/sql/repository/pipelineConfig/CdWorfkflowRepository.go

@@ -100,6 +100,9 @@ const (
 	WorkflowSucceeded          = "Succeeded"
 	WorkflowTimedOut           = "TimedOut"
 	WorkflowUnableToFetchState = "UnableToFetch"
+	WorkflowTypeDeploy         = "DEPLOY"
+	WorkflowTypePre            = "PRE"
+	WorkflowTypePost           = "POST"
 )
 
 func (a WorkflowStatus) String() string {
@@ -166,6 +169,16 @@ type CdWorkflowRunner struct {
 	sql.AuditLog
 }
 
+func (c *CdWorkflowRunner) IsExternalRun() bool {
+	var isExtCluster bool
+	if c.WorkflowType == WorkflowTypePre {
+		isExtCluster = c.CdWorkflow.Pipeline.RunPreStageInEnv
+	} else if c.WorkflowType == WorkflowTypePost {
+		isExtCluster = c.CdWorkflow.Pipeline.RunPostStageInEnv
+	}
+	return isExtCluster
+}
+
 type CiPipelineMaterialResponse struct {
 	Id              int                    `json:"id"`
 	GitMaterialId   int                    `json:"gitMaterialId"`

internal/sql/repository/pipelineConfig/CiWorkflowRepository.go

@@ -72,9 +72,13 @@ type CiWorkflow struct {
 	CiBuildType            string            `sql:"ci_build_type"`
 	EnvironmentId          int               `sql:"environment_id"`
 	ImagePathReservationId int               `sql:"image_path_reservation_id"`
-	ReferenceCiWorkflowId int               `sql:"ref_ci_workflow_id"`
-	ParentCiWorkFlowId    int               `sql:"parent_ci_workflow_id"`
-	CiPipeline            *CiPipeline
+	ReferenceCiWorkflowId  int               `sql:"ref_ci_workflow_id"`
+	ParentCiWorkFlowId     int               `sql:"parent_ci_workflow_id"`
+	CiPipeline             *CiPipeline
+}
+
+func (r *CiWorkflow) IsExternalRunInJobType() bool {
+	return r.EnvironmentId != 0
 }
 
 type WorkflowWithArtifact struct {
@@ -101,8 +105,8 @@ type WorkflowWithArtifact struct {
 	EnvironmentId          int               `json:"environmentId"`
 	EnvironmentName        string            `json:"environmentName"`
 	ImagePathReservationId int               `json:"image_path_reservation_id"`
-	RefCiWorkflowId    int               `json:"referenceCiWorkflowId"`
-	ParentCiWorkflowId int               `json:"parent_ci_workflow_id"`
+	RefCiWorkflowId        int               `json:"referenceCiWorkflowId"`
+	ParentCiWorkflowId     int               `json:"parent_ci_workflow_id"`
 }
 
 type GitCommit struct {

pkg/cluster/ClusterService.go

@@ -180,6 +180,9 @@ type ClusterService interface {
 	ConnectClustersInBatch(clusters []*ClusterBean, clusterExistInDb bool)
 	ConvertClusterBeanToCluster(clusterBean *ClusterBean, userId int32) *repository.Cluster
 	ConvertClusterBeanObjectToCluster(bean *ClusterBean) *v1alpha1.Cluster
+
+	GetClusterConfigByClusterId(clusterId int) (*k8s.ClusterConfig, error)
+	GetClusterConfigByEnvId(envId int) (*k8s.ClusterConfig, error)
 }
 
 type ClusterServiceImpl struct {
@@ -1095,3 +1098,32 @@ func (impl ClusterServiceImpl) ConvertClusterBeanObjectToCluster(bean *ClusterBe
 	}
 	return cl
 }
+
+func (impl ClusterServiceImpl) GetClusterConfigByClusterId(clusterId int) (*k8s.ClusterConfig, error) {
+	clusterBean, err := impl.FindById(clusterId)
+	if err != nil {
+		impl.logger.Errorw("error in getting clusterBean by cluster id", "err", err, "clusterId", clusterId)
+		return nil, err
+	}
+	rq := *clusterBean
+	clusterConfig, err := rq.GetClusterConfig()
+	if err != nil {
+		impl.logger.Errorw("error in getting cluster config", "err", err, "clusterId", clusterBean.Id)
+		return nil, err
+	}
+	return clusterConfig, nil
+}
+
+func (impl ClusterServiceImpl) GetClusterConfigByEnvId(envId int) (*k8s.ClusterConfig, error) {
+	envBean, err := impl.environmentService.FindById(envId)
+	if err != nil {
+		impl.logger.Errorw("error in getting envBean by envId", "err", err, "envId", envId)
+		return nil, err
+	}
+	clusterConfig, err := impl.GetClusterConfigByClusterId(envBean.ClusterId)
+	if err != nil {
+		impl.logger.Errorw("error in getting cluster config by env id", "err", err, "envId", envId)
+		return nil, err
+	}
+	return clusterConfig, nil
+}

pkg/generateManifest/DeployementTemplateService_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	client2 "github.com/devtron-labs/authenticator/client"
+	"github.com/devtron-labs/common-lib/utils/k8s"
 	"github.com/devtron-labs/devtron/api/bean"
 	client "github.com/devtron-labs/devtron/api/helm-app"
 	mocks4 "github.com/devtron-labs/devtron/api/helm-app/mocks"
@@ -16,7 +17,6 @@ import (
 	"github.com/devtron-labs/devtron/pkg/chart/mocks"
 	chartRepoRepository "github.com/devtron-labs/devtron/pkg/chartRepo/repository"
 	mocks5 "github.com/devtron-labs/devtron/pkg/chartRepo/repository/mocks"
-	"github.com/devtron-labs/devtron/util/k8s"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"testing"

pkg/pipeline/BlobStorageConfigService.go

@@ -0,0 +1,107 @@
+package pipeline
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	blob_storage "github.com/devtron-labs/common-lib/blob-storage"
+	"github.com/devtron-labs/common-lib/utils/k8s"
+	bean2 "github.com/devtron-labs/devtron/pkg/pipeline/bean"
+	"go.uber.org/zap"
+	v12 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"strconv"
+)
+
+type BlobStorageConfigService interface {
+	FetchCmAndSecretBlobConfigFromExternalCluster(clusterConfig *k8s.ClusterConfig, namespace string) (*bean2.CmBlobStorageConfig, *bean2.SecretBlobStorageConfig, error)
+}
+type BlobStorageConfigServiceImpl struct {
+	Logger     *zap.SugaredLogger
+	k8sUtil    *k8s.K8sUtil
+	ciCdConfig *CiCdConfig
+}
+
+func NewBlobStorageConfigServiceImpl(Logger *zap.SugaredLogger, k8sUtil *k8s.K8sUtil, ciCdConfig *CiCdConfig) *BlobStorageConfigServiceImpl {
+	return &BlobStorageConfigServiceImpl{
+		Logger:     Logger,
+		k8sUtil:    k8sUtil,
+		ciCdConfig: ciCdConfig,
+	}
+}
+
+func (impl *BlobStorageConfigServiceImpl) FetchCmAndSecretBlobConfigFromExternalCluster(clusterConfig *k8s.ClusterConfig, namespace string) (*bean2.CmBlobStorageConfig, *bean2.SecretBlobStorageConfig, error) {
+	cmConfig := &bean2.CmBlobStorageConfig{}
+	secretConfig := &bean2.SecretBlobStorageConfig{}
+	_, _, kubeClient, err := impl.k8sUtil.GetK8sConfigAndClients(&k8s.ClusterConfig{})
+	if err != nil {
+		impl.Logger.Errorw("FetchCmAndSecretBlobConfigFromExternalCluster, error in getting kubeClient by cluster config", "err", err)
+		return cmConfig, secretConfig, err
+	}
+	cv1 := kubeClient.CoreV1()
+	ctx := context.Background()
+	opts := v12.GetOptions{}
+	cmName := impl.ciCdConfig.ExtBlobStorageCmName
+	secretName := impl.ciCdConfig.ExtBlobStorageSecretName
+	cm, err := cv1.ConfigMaps(namespace).Get(ctx, cmName, opts)
+	if err != nil {
+		impl.Logger.Errorw("error in getting config map in external cluster", "err", err, "blobStorageCmName", impl.ciCdConfig.ExtBlobStorageCmName, "clusterName", clusterConfig.ClusterName)
+		return cmConfig, secretConfig, err
+	}
+	secret, err := cv1.Secrets(namespace).Get(ctx, secretName, opts)
+	if err != nil {
+		impl.Logger.Errorw("error in getting secret in external cluster", "err", err, "blobStorageSecretName", impl.ciCdConfig.ExtBlobStorageSecretName, "clusterName", clusterConfig.ClusterName)
+		return cmConfig, secretConfig, err
+	}
+	//for IAM configured in S3 in external cluster, get logs/artifact will not work
+	if cm.Data != nil && secret.Data != nil {
+		err = cmConfig.SetCmBlobStorageConfig(cm.Data)
+		if err != nil {
+			fmt.Println("error marshalling external blob storage cm data to struct:", err)
+			return cmConfig, secretConfig, err
+		}
+		err = secretConfig.SetSecretBlobStorageConfig(secret.Data)
+		if err != nil {
+			fmt.Println("error marshalling external blob storage secret data to struct:", err)
+			return cmConfig, secretConfig, err
+		}
+	}
+	if cm.Data == nil {
+		fmt.Println("Data field not found in config map")
+	}
+	if secret.Data == nil {
+		fmt.Println("Data field not found in secret")
+	}
+	impl.Logger.Infow("fetching cm and secret from external cluster cloud provider", "ext cluster config: ", cmConfig)
+	return cmConfig, secretConfig, nil
+}
+
+func updateRequestWithExtClusterCmAndSecret(request *blob_storage.BlobStorageRequest, cmConfig *bean2.CmBlobStorageConfig, secretConfig *bean2.SecretBlobStorageConfig) *blob_storage.BlobStorageRequest {
+	request.StorageType = cmConfig.CloudProvider
+
+	request.AwsS3BaseConfig.AccessKey = cmConfig.S3AccessKey
+	request.AwsS3BaseConfig.EndpointUrl = cmConfig.S3Endpoint
+	request.AwsS3BaseConfig.Passkey = decodeSecretKey(secretConfig.S3SecretKey)
+	isInSecure, _ := strconv.ParseBool(cmConfig.S3EndpointInsecure)
+	request.AwsS3BaseConfig.IsInSecure = isInSecure
+	request.AwsS3BaseConfig.BucketName = cmConfig.CdDefaultBuildLogsBucket
+	request.AwsS3BaseConfig.Region = cmConfig.CdDefaultCdLogsBucketRegion
+	s3BucketVersioned, _ := strconv.ParseBool(cmConfig.S3BucketVersioned)
+	request.AwsS3BaseConfig.VersioningEnabled = s3BucketVersioned
+
+	request.AzureBlobBaseConfig.AccountName = cmConfig.AzureAccountName
+	request.AzureBlobBaseConfig.AccountKey = decodeSecretKey(secretConfig.AzureAccountKey)
+	request.AzureBlobBaseConfig.BlobContainerName = cmConfig.AzureBlobContainerCiLog
+
+	request.GcpBlobBaseConfig.CredentialFileJsonData = decodeSecretKey(secretConfig.GcpBlobStorageCredentialJson)
+	request.GcpBlobBaseConfig.BucketName = cmConfig.CdDefaultBuildLogsBucket
+
+	return request
+}
+
+func decodeSecretKey(secretKey string) string {
+	decodedKey, err := base64.StdEncoding.DecodeString(secretKey)
+	if err != nil {
+		fmt.Println("error decoding base64 key:", err)
+	}
+	return string(decodedKey)
+}

pkg/pipeline/CdHandler.go

@@ -119,9 +119,11 @@ type CdHandlerImpl struct {
 	k8sUtil                                *k8s.K8sUtil
 	workflowService                        WorkflowService
 	config                                 *CdConfig
+	clusterService                         cluster.ClusterService
+	blobConfigStorageService               BlobStorageConfigService
 }
 
-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, clusterService cluster.ClusterService, blobConfigStorageService BlobStorageConfigService) *CdHandlerImpl {
 	cdh := &CdHandlerImpl{
 		Logger:                                 Logger,
 		userService:                            userService,
@@ -155,6 +157,8 @@ func NewCdHandlerImpl(Logger *zap.SugaredLogger, userService user.UserService, c
 		imageTaggingService:                    imageTaggingService,
 		k8sUtil:                                k8sUtil,
 		workflowService:                        workflowService,
+		clusterService:                         clusterService,
+		blobConfigStorageService:               blobConfigStorageService,
 	}
 	config, err := GetCdConfig()
 	if err != nil {
@@ -913,7 +917,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,8 +926,8 @@ 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) {
-	impl.Logger.Debug("getting historic logs")
+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", "pipelineId", pipelineId)
 
 	cdConfig, err := impl.cdWorkflowRepository.FindConfigByPipelineId(pipelineId)
 	if err != nil && !util.IsErrNoRows(err) {
@@ -964,6 +968,19 @@ func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pip
 			CredentialFileJsonData: impl.config.BlobStorageGcpCredentialJson,
 		},
 	}
+	useExternalBlobStorage := isExternalBlobStorageEnabled(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.blobConfigStorageService.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
+		}
+		rq := &cdLogRequest
+		rq.SetBuildLogRequest(cmConfig, secretConfig)
+	}
+
 	impl.Logger.Infow("s3 log req ", "req", cdLogRequest)
 	oldLogsStream, cleanUp, err := impl.ciLogService.FetchLogs(impl.config.BaseLogLocationPath, cdLogRequest)
 	if err != nil {
@@ -973,6 +990,10 @@ func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pip
 	logReader := bufio.NewReader(oldLogsStream)
 	return logReader, cleanUp, err
 }
+func isExternalBlobStorageEnabled(isExternalRun bool, useBlobStorageConfigInCdWorkflow bool) bool {
+	//TODO impl.config.UseBlobStorageConfigInCdWorkflow fetches the live status, we need to check from db as well, we should put useExternalBlobStorage in db
+	return isExternalRun && !useBlobStorageConfigInCdWorkflow
+}
 
 func (impl *CdHandlerImpl) FetchCdWorkflowDetails(appId int, environmentId int, pipelineId int, buildId int) (WorkflowResponse, error) {
 	workflowR, err := impl.cdWorkflowRepository.FindWorkflowRunnerById(buildId)
@@ -1065,7 +1086,7 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 		impl.Logger.Errorw("unable to fetch ciWorkflow", "err", err)
 		return nil, err
 	}
-
+	useExternalBlobStorage := isExternalBlobStorageEnabled(wfr.IsExternalRun(), impl.config.UseBlobStorageConfigInCdWorkflow)
 	if !wfr.BlobStorageEnabled {
 		return nil, errors.New("logs-not-stored-in-repository")
 	}
@@ -1115,6 +1136,21 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 		AwsS3BaseConfig:     awsS3BaseConfig,
 		GcpBlobBaseConfig:   gcpBlobBaseConfig,
 	}
+	if useExternalBlobStorage {
+		clusterConfig, err := impl.clusterService.GetClusterConfigByClusterId(wfr.CdWorkflow.Pipeline.Environment.ClusterId)
+		if err != nil {
+			impl.Logger.Errorw("GetClusterConfigByClusterId, error in fetching clusterConfig", "err", err, "clusterId", wfr.CdWorkflow.Pipeline.Environment.ClusterId)
+			return nil, err
+		}
+		//fetch extClusterBlob cm and cs from k8s client, if they are present then read creds
+		//from them else return.
+		cmConfig, secretConfig, err := impl.blobConfigStorageService.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 = updateRequestWithExtClusterCmAndSecret(request, cmConfig, secretConfig)
+	}
 	_, 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 {