func()

in packages/cli/internal/pkg/cli/logs_workflow.go [77:136]

• 55 lines of code
• 14 McCabe index (conditional complexity)

func (o *logsWorkflowOpts) Execute() error {
	err := o.setRunId()
	if err != nil {
		if errors.Is(err, noRunsFoundError) {
			log.Info().Msgf("Workflow '%s' has not been run yet", o.workflowName)
			return nil
		}
		return err
	}
	log.Debug().Msgf("Showing logs for workflow run '%s'", o.runId)

	runLog, err := o.workflowManager.GetRunLog(o.runId)
	if err != nil {
		return err
	}

	var jobIds []string
	if o.taskId != "" {
		if !containsTaskId(o.taskId, runLog.Tasks) {
			log.Info().Msgf("Task `%s` does not exist for run `%s`", o.taskId, o.runId)
			return nil
		}
		jobIds = []string{o.taskId}
	} else if o.allTasks || o.failedTasks {
		jobIds, err = o.getJobIds(runLog.Tasks)
		if err != nil {
			return err
		}
	} else {
		printRunLog(runLog)
		return nil
	}

	if len(jobIds) == 0 {
		log.Info().Msgf("No logs available for run '%s'. Please try again later.", o.runId)
		return nil
	}
	notCachedJobIds := filterCachedJobIds(jobIds)
	totalJobs := len(jobIds)
	notCachedJobs := len(notCachedJobIds)
	cachedJobs := totalJobs - notCachedJobs
	if cachedJobs > 0 {
		log.Info().Msgf("%d of %d jobs were cached. Logs are not available for cached jobs", cachedJobs, totalJobs)
	}
	if notCachedJobs == 0 {
		return nil
	}

	streamNames, err := o.getStreamsForJobs(notCachedJobIds)
	if err != nil {
		return err
	}

	logGroupName := "/aws/batch/job"
	if o.tail {
		return o.followLogStreams(logGroupName, streamNames...)
	} else {
		return o.displayLogStreams(logGroupName, o.startTime, o.endTime, o.filter, streamNames...)
	}
}