Skip to content

feat: wf logs #6606

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 23, 2025
Merged

feat: wf logs #6606

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
675255e
wf logs
nishant-d May 23, 2025
c5a9787
Merge branch 'develop' into wf-log-follow
nishant-d May 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion api/restHandler/app/pipeline/configure/BuildPipelineRestHandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1085,6 +1085,16 @@ func (handler *PipelineConfigRestHandlerImpl) GetBuildLogs(w http.ResponseWriter
common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
return
}
followLogs := true
if ok := r.URL.Query().Has("followLogs"); ok {
followLogsStr := r.URL.Query().Get("followLogs")
follow, err := strconv.ParseBool(followLogsStr)
if err != nil {
common.WriteJsonResp(w, err, "followLogs is not a valid bool", http.StatusBadRequest)
return
}
followLogs = follow
}

workflowId, err := strconv.Atoi(vars["workflowId"])
if err != nil {
Expand Down Expand Up @@ -1116,7 +1126,7 @@ func (handler *PipelineConfigRestHandlerImpl) GetBuildLogs(w http.ResponseWriter
return
}
}
logsReader, cleanUp, err := handler.ciHandlerService.GetRunningWorkflowLogs(workflowId)
logsReader, cleanUp, err := handler.ciHandlerService.GetRunningWorkflowLogs(workflowId, followLogs)
if err != nil {
handler.Logger.Errorw("service err, GetBuildLogs", "err", err, "pipelineId", pipelineId, "workflowId", workflowId, "lastEventId", lastEventId)
common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
Expand Down
12 changes: 11 additions & 1 deletion api/restHandler/app/pipeline/configure/DeploymentPipelineRestHandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1609,6 +1609,16 @@ func (handler *PipelineConfigRestHandlerImpl) GetPrePostDeploymentLogs(w http.Re
common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
return
}
followLogs := true
if ok := r.URL.Query().Has("followLogs"); ok {
followLogsStr := r.URL.Query().Get("followLogs")
follow, err := strconv.ParseBool(followLogsStr)
if err != nil {
common.WriteJsonResp(w, err, "followLogs is not a valid bool", http.StatusBadRequest)
return
}
followLogs = follow
}
handler.Logger.Infow("request payload, GetPrePostDeploymentLogs", "err", err, "appId", appId, "environmentId", environmentId, "pipelineId", pipelineId, "workflowId", workflowId)

// RBAC CHECK
Expand All @@ -1619,7 +1629,7 @@ func (handler *PipelineConfigRestHandlerImpl) GetPrePostDeploymentLogs(w http.Re
}
// RBAC CHECK

logsReader, cleanUp, err := handler.cdHandlerService.GetRunningWorkflowLogs(environmentId, pipelineId, workflowId)
logsReader, cleanUp, err := handler.cdHandlerService.GetRunningWorkflowLogs(environmentId, pipelineId, workflowId, followLogs)
if err != nil {
handler.Logger.Errorw("service err, GetPrePostDeploymentLogs", "err", err, "appId", appId, "environmentId", environmentId, "pipelineId", pipelineId, "workflowId", workflowId)
common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
Expand Down
10 changes: 5 additions & 5 deletions pkg/build/trigger/HandlerService.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ type HandlerService interface {
StartCiWorkflowAndPrepareWfRequest(trigger types.Trigger) (*pipelineConfig.CiPipeline, map[string]string, *pipelineConfig.CiWorkflow, *types.WorkflowRequest, error)

CancelBuild(workflowId int, forceAbort bool) (int, error)
GetRunningWorkflowLogs(workflowId int) (*bufio.Reader, func() error, error)
GetRunningWorkflowLogs(workflowId int, followLogs bool) (*bufio.Reader, func() error, error)
GetHistoricBuildLogs(workflowId int, ciWorkflow *pipelineConfig.CiWorkflow) (map[string]string, error)
DownloadCiWorkflowArtifacts(pipelineId int, buildId int) (*os.File, error)
}
Expand Down Expand Up @@ -1685,16 +1685,16 @@ func (impl *HandlerServiceImpl) getRestConfig(workflow *pipelineConfig.CiWorkflo
return restConfig, nil
}

func (impl *HandlerServiceImpl) GetRunningWorkflowLogs(workflowId int) (*bufio.Reader, func() error, error) {
func (impl *HandlerServiceImpl) GetRunningWorkflowLogs(workflowId int, followLogs bool) (*bufio.Reader, func() error, error) {
ciWorkflow, err := impl.ciWorkflowRepository.FindById(workflowId)
if err != nil {
impl.Logger.Errorw("err", "err", err)
return nil, nil, err
}
return impl.getWorkflowLogs(ciWorkflow)
return impl.getWorkflowLogs(ciWorkflow, followLogs)
}

func (impl *HandlerServiceImpl) getWorkflowLogs(ciWorkflow *pipelineConfig.CiWorkflow) (*bufio.Reader, func() error, error) {
func (impl *HandlerServiceImpl) getWorkflowLogs(ciWorkflow *pipelineConfig.CiWorkflow, followLogs bool) (*bufio.Reader, func() error, error) {
if string(v1alpha1.NodePending) == ciWorkflow.PodStatus {
return bufio.NewReader(strings.NewReader("")), func() error { return nil }, nil
}
Expand All @@ -1717,7 +1717,7 @@ func (impl *HandlerServiceImpl) getWorkflowLogs(ciWorkflow *pipelineConfig.CiWor
isExt = true
}

logStream, cleanUp, err := impl.ciLogService.FetchRunningWorkflowLogs(ciLogRequest, clusterConfig, isExt)
logStream, cleanUp, err := impl.ciLogService.FetchRunningWorkflowLogs(ciLogRequest, clusterConfig, isExt, followLogs)
if logStream == nil || err != nil {
if !ciWorkflow.BlobStorageEnabled {
return nil, nil, &util.ApiError{Code: "200", HttpStatusCode: 400, UserMessage: "logs-not-stored-in-repository"}
Expand Down
2 changes: 1 addition & 1 deletion pkg/deployment/trigger/devtronApps/HandlerService.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ type HandlerService interface {

CancelStage(workflowRunnerId int, forceAbort bool, userId int32) (int, error)
DownloadCdWorkflowArtifacts(buildId int) (*os.File, error)
GetRunningWorkflowLogs(environmentId int, pipelineId int, workflowId int) (*bufio.Reader, func() error, error)
GetRunningWorkflowLogs(environmentId int, pipelineId int, workflowId int, followLogs bool) (*bufio.Reader, func() error, error)
}

type HandlerServiceImpl struct {
Expand Down
8 changes: 4 additions & 4 deletions pkg/deployment/trigger/devtronApps/prePostWfAndLogsCode.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,7 @@ func (impl *HandlerServiceImpl) DownloadCdWorkflowArtifacts(buildId int) (*os.Fi
return file, nil
}

func (impl *HandlerServiceImpl) GetRunningWorkflowLogs(environmentId int, pipelineId int, wfrId int) (*bufio.Reader, func() error, error) {
func (impl *HandlerServiceImpl) GetRunningWorkflowLogs(environmentId int, pipelineId int, wfrId int, followLogs bool) (*bufio.Reader, func() error, error) {
cdWorkflow, err := impl.cdWorkflowRepository.FindWorkflowRunnerById(wfrId)
if err != nil {
impl.logger.Errorw("error on fetch wf runner", "err", err)
Expand Down Expand Up @@ -253,16 +253,16 @@ func (impl *HandlerServiceImpl) GetRunningWorkflowLogs(environmentId int, pipeli
} else if cdWorkflow.WorkflowType == types.POST {
isExtCluster = pipeline.RunPostStageInEnv
}
return impl.getWorkflowLogs(pipelineId, cdWorkflow, clusterConfig, isExtCluster)
return impl.getWorkflowLogs(pipelineId, cdWorkflow, clusterConfig, isExtCluster, followLogs)
}

func (impl *HandlerServiceImpl) getWorkflowLogs(pipelineId int, cdWorkflow *pipelineConfig.CdWorkflowRunner, clusterConfig *k8s.ClusterConfig, runStageInEnv bool) (*bufio.Reader, func() error, error) {
func (impl *HandlerServiceImpl) getWorkflowLogs(pipelineId int, cdWorkflow *pipelineConfig.CdWorkflowRunner, clusterConfig *k8s.ClusterConfig, runStageInEnv bool, followLogs bool) (*bufio.Reader, func() error, error) {
cdLogRequest := types.BuildLogRequest{
PodName: cdWorkflow.PodName,
Namespace: cdWorkflow.Namespace,
}

logStream, cleanUp, err := impl.ciLogService.FetchRunningWorkflowLogs(cdLogRequest, clusterConfig, runStageInEnv)
logStream, cleanUp, err := impl.ciLogService.FetchRunningWorkflowLogs(cdLogRequest, clusterConfig, runStageInEnv, followLogs)
if logStream == nil || err != nil {
if !cdWorkflow.BlobStorageEnabled {
return nil, nil, errors.New("logs-not-stored-in-repository")
Expand Down
6 changes: 3 additions & 3 deletions pkg/pipeline/CiLogService.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import (
)

type CiLogService interface {
FetchRunningWorkflowLogs(ciLogRequest types.BuildLogRequest, clusterConfig *k8s.ClusterConfig, isExt bool) (io.ReadCloser, func() error, error)
FetchRunningWorkflowLogs(ciLogRequest types.BuildLogRequest, clusterConfig *k8s.ClusterConfig, isExt bool, followLogs bool) (io.ReadCloser, func() error, error)
FetchLogs(baseLogLocationPathConfig string, ciLogRequest types.BuildLogRequest) (*os.File, func() error, error)
}

Expand All @@ -53,7 +53,7 @@ func NewCiLogServiceImpl(logger *zap.SugaredLogger, k8sUtil *k8s.K8sServiceImpl)
}, nil
}

func (impl *CiLogServiceImpl) FetchRunningWorkflowLogs(ciLogRequest types.BuildLogRequest, clusterConfig *k8s.ClusterConfig, isExt bool) (io.ReadCloser, func() error, error) {
func (impl *CiLogServiceImpl) FetchRunningWorkflowLogs(ciLogRequest types.BuildLogRequest, clusterConfig *k8s.ClusterConfig, isExt bool, followLogs bool) (io.ReadCloser, func() error, error) {
var kubeClient *kubernetes.Clientset
kubeClient = impl.kubeClient
var err error
Expand All @@ -64,7 +64,7 @@ func (impl *CiLogServiceImpl) FetchRunningWorkflowLogs(ciLogRequest types.BuildL
return nil, nil, err
}
}
req := impl.k8sUtil.GetLogsForAPod(kubeClient, ciLogRequest.Namespace, ciLogRequest.PodName, bean.Main, true)
req := impl.k8sUtil.GetLogsForAPod(kubeClient, ciLogRequest.Namespace, ciLogRequest.PodName, bean.Main, followLogs)
podLogs, err := req.Stream(context.Background())
if err != nil {
impl.logger.Errorw("error in opening stream", "name", ciLogRequest.PodName, "err", err)
Expand Down
Loading