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

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/CdHandler.go

@@ -20,6 +20,7 @@ package pipeline
 import (
 	"bufio"
 	"context"
+	"encoding/base64"
 	"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 {
@@ -913,7 +918,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 +927,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 +969,19 @@ 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 = assignNewBlobStorageConfigInCdLogRequest(&cdLogRequest, 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 +991,75 @@ func (impl *CdHandlerImpl) getLogsFromRepository(pipelineId int, cdWorkflow *pip
 	logReader := bufio.NewReader(oldLogsStream)
 	return logReader, cleanUp, err
 }
+func assignNewBlobStorageConfigInCdLogRequest(cdLogRequest *BuildLogRequest, cmConfig *bean2.CmBlobStorageConfig, secretConfig *bean2.SecretBlobStorageConfig) BuildLogRequest {
+	cdLogRequest.CloudProvider = cmConfig.CloudProvider
+	cdLogRequest.AzureBlobConfig.AccountName = cmConfig.AzureAccountName
+	cdLogRequest.AzureBlobConfig.AccountKey = decodeSecretKey(secretConfig.AzureAccountKey)
+	cdLogRequest.AzureBlobConfig.BlobContainerName = cmConfig.AzureBlobContainerCiLog
+
+	cdLogRequest.GcpBlobBaseConfig.CredentialFileJsonData = decodeSecretKey(secretConfig.GcpBlobStorageCredentialJson)
+	cdLogRequest.GcpBlobBaseConfig.BucketName = cmConfig.CdDefaultBuildLogsBucket
+
+	cdLogRequest.AwsS3BaseConfig.AccessKey = cmConfig.S3AccessKey
+	cdLogRequest.AwsS3BaseConfig.EndpointUrl = cmConfig.S3Endpoint
+	cdLogRequest.AwsS3BaseConfig.Passkey = decodeSecretKey(secretConfig.S3SecretKey)
+	isEndpointInSecure, _ := strconv.ParseBool(cmConfig.S3EndpointInsecure)
+	cdLogRequest.AwsS3BaseConfig.IsInSecure = isEndpointInSecure
+	cdLogRequest.AwsS3BaseConfig.BucketName = cmConfig.CdDefaultBuildLogsBucket
+	cdLogRequest.AwsS3BaseConfig.Region = cmConfig.CdDefaultCdLogsBucketRegion
+	s3BucketVersioned, _ := strconv.ParseBool(cmConfig.S3BucketVersioned)
+	cdLogRequest.AwsS3BaseConfig.VersioningEnabled = s3BucketVersioned
+
+	return *cdLogRequest
+}
+
+func (impl *CdHandlerImpl) FetchCmAndSecretBlobConfigFromExternalCluster(clusterConfig *k8s.ClusterConfig, namespace string) (*bean2.CmBlobStorageConfig, *bean2.SecretBlobStorageConfig, error) {
+	cmConfig := &bean2.CmBlobStorageConfig{}
+	secretConfig := &bean2.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 {
+		err = cmConfig.PopulateWithK8sExtBlobCmData(cm.Data)
+		if err != nil {
+			fmt.Println("error marshalling external blob storage cm data to struct:", err)
+			return cmConfig, secretConfig, err
+		}
+		err = secretConfig.PopulateWithK8sExtBlobSecretData(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 {
+		return cmConfig, secretConfig, errors.New("Data field not found in config map")
+	}
+	if secret.Data == nil {
+		return cmConfig, secretConfig, errors.New("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 (impl *CdHandlerImpl) FetchCdWorkflowDetails(appId int, environmentId int, pipelineId int, buildId int) (WorkflowResponse, error) {
 	workflowR, err := impl.cdWorkflowRepository.FindWorkflowRunnerById(buildId)
@@ -1065,7 +1152,14 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 		impl.Logger.Errorw("unable to fetch ciWorkflow", "err", err)
 		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
+	}
+	//TODO impl.config.UseBlobStorageConfigInCdWorkflow fetches the live status, we need to check from db as well, we should put useExternalBlobStorage in db as well
+	useExternalBlobStorage := isExtCluster && !impl.config.UseBlobStorageConfigInCdWorkflow
 	if !wfr.BlobStorageEnabled {
 		return nil, errors.New("logs-not-stored-in-repository")
 	}
@@ -1115,6 +1209,30 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 		AwsS3BaseConfig:     awsS3BaseConfig,
 		GcpBlobBaseConfig:   gcpBlobBaseConfig,
 	}
+	if useExternalBlobStorage {
+		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
+		}
+		//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 = assignNewBlobStorageConfigInRequest(request, cmConfig, secretConfig)
+	}
 	_, numBytes, err := blobStorageService.Get(request)
 	if err != nil {
 		impl.Logger.Errorw("error occurred while downloading file", "request", request, "error", err)
@@ -1131,6 +1249,37 @@ func (impl *CdHandlerImpl) DownloadCdWorkflowArtifacts(pipelineId int, buildId i
 	return file, nil
 }
 
+func assignNewBlobStorageConfigInRequest(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)
+}
+
 func (impl *CdHandlerImpl) converterWFR(wfr pipelineConfig.CdWorkflowRunner) pipelineConfig.CdWorkflowWithArtifact {
 	workflow := pipelineConfig.CdWorkflowWithArtifact{}
 	if wfr.Id > 0 {

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

@@ -243,6 +243,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
 }