Introducing Workspaces into pipelines

pipelines/manager.go

@@ -136,3 +136,85 @@ func (sm *PipelinesServicesManager) CancelRun(runID int) error {
 	runService.ServiceDetails = sm.config.GetServiceDetails()
 	return runService.CancelRun(runID)
 }
+
+func (sm *PipelinesServicesManager) ValidatePipelineSources(data []byte) error {
+	validateService := services.NewValidateService(sm.client)
+	validateService.ServiceDetails = sm.config.GetServiceDetails()
+	return validateService.ValidatePipeline(data)
+}
+
+func (sm *PipelinesServicesManager) WorkspaceSync(projectKey string) error {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.WorkspaceSync(projectKey)
+}
+
+func (sm *PipelinesServicesManager) WorkspacePollSyncStatus() ([]services.WorkspacesResponse, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.WorkspacePollSyncStatus()
+}
+
+func (sm *PipelinesServicesManager) WorkspacePipelines() (map[string]string, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	workspacesResponseStatus, err := sm.WorkspacePollSyncStatus()
+	if err != nil {
+		return nil, err
+	}
+	return workspaceService.GetWorkspacePipelines(workspacesResponseStatus)
+}
+
+func (sm *PipelinesServicesManager) WorkspaceTriggerPipelines(branch, pipName string, isMultiBranch bool) error {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return sm.TriggerPipelineRun(branch, pipName, isMultiBranch)
+}
+
+func (sm *PipelinesServicesManager) WorkspaceRunIDs(pipelineNames []string) ([]services.PipelinesRunID, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.WorkspaceRunIDs(pipelineNames)
+}
+
+func (sm *PipelinesServicesManager) WorkspaceRunStatus(pipeRunID int) ([]byte, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.WorkspaceRunStatus(pipeRunID)
+}
+
+func (sm *PipelinesServicesManager) WorkspaceStepStatus(pipeRunID int) ([]byte, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.WorkspaceStepStatus(pipeRunID)
+}
+
+func (sm *PipelinesServicesManager) ValidateWorkspace(data []byte) error {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.ValidateWorkspace(data)
+}
+
+func (sm *PipelinesServicesManager) GetStepConsoles(stepID string) (map[string][]services.Console, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.GetStepLogsUsingStepID(stepID)
+}
+
+func (sm *PipelinesServicesManager) GetStepletConsoles(stepID string) (map[string][]services.Console, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.GetStepletLogsUsingStepID(stepID)
+}
+
+func (sm *PipelinesServicesManager) GetWorkspace() ([]services.WorkspacesResponse, error) {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.GetWorkspace()
+}
+
+func (sm *PipelinesServicesManager) DeleteWorkspace(projectKey string) error {
+	workspaceService := services.NewWorkspaceService(sm.client)
+	workspaceService.ServiceDetails = sm.config.GetServiceDetails()
+	return workspaceService.DeleteWorkspace(projectKey)
+}

pipelines/services/run.go

@@ -13,6 +13,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 )
 
 type RunService struct {
@@ -99,19 +100,29 @@ func (rs *RunService) TriggerPipelineRun(branch, pipeline string, isMultiBranch
 		return err
 	}
 
-	// Prepare Request
-	resp, body, err := rs.client.SendPost(uri, buf.Bytes(), &httpDetails)
-	if err != nil {
-		return err
+	pollingAction := func() (shouldStop bool, responseBody []byte, err error) {
+		// Prepare Request
+		resp, body, err := rs.client.SendPost(uri, buf.Bytes(), &httpDetails)
+		if err != nil {
+			return true, body, err
+		}
+
+		// Response Analysis
+		if err := errorutils.CheckResponseStatusWithBody(resp, body, http.StatusOK); err != nil {
+			return false, body, err
+		}
+		log.Info(fmt.Sprintf("Triggered successfully\n%s %s \n%14s %s", "PipelineName :", pipeline, "Branch :", branch))
+		return true, body, err
 	}
-
-	// Response Analysis
-	if err := errorutils.CheckResponseStatusWithBody(resp, body, http.StatusOK); err != nil {
-		return err
+	pollingExecutor := &httputils.PollingExecutor{
+		Timeout:         2 * time.Minute,
+		PollingInterval: 5 * time.Second,
+		PollingAction:   pollingAction,
+		MsgPrefix:       "Get pipeline workspace sync status...",
 	}
-	log.Info(fmt.Sprintf("Triggered successfully\n%s %s \n%14s %s", "PipelineName :", pipeline, "Branch :", branch))
-
-	return nil
+	// Polling execution
+	_, err = pollingExecutor.Execute()
+	return err
 }
 
 func (rs *RunService) CancelRun(runID int) error {

pipelines/services/types.go

@@ -87,3 +87,71 @@ type PipelineResources struct {
 	CreatedAt                time.Time   `json:"createdAt,omitempty"`
 	UpdatedAt                time.Time   `json:"updatedAt,omitempty"`
 }
+
+type WorkspacesResponse struct {
+	ValuesYmlPropertyBag    interface{}             `json:"valuesYmlPropertyBag,omitempty"`
+	PipelinesYmlPropertyBag PipelinesYmlPropertyBag `json:"pipelinesYmlPropertyBag,omitempty"`
+	ID                      int                     `json:"id,omitempty"`
+	Name                    string                  `json:"name,omitempty"`
+	ProjectID               int                     `json:"projectId,omitempty"`
+	IsSyncing               *bool                   `json:"isSyncing,omitempty"`
+	LastSyncStatusCode      int                     `json:"lastSyncStatusCode,omitempty"`
+	LastSyncStartedAt       time.Time               `json:"lastSyncStartedAt,omitempty"`
+	LastSyncEndedAt         interface{}             `json:"lastSyncEndedAt,omitempty"`
+	LastSyncLogs            string                  `json:"lastSyncLogs,omitempty"`
+	CreatedBy               int                     `json:"createdBy,omitempty"`
+}
+
+type PipelinesYmlPropertyBag struct {
+	Resources []Resources `json:"resources,omitempty"`
+	Pipelines []Pipelines `json:"pipelines,omitempty"`
+}
+
+type Resources struct {
+	Configuration Configuration `json:"configuration,omitempty"`
+	Name          string        `json:"name,omitempty"`
+	Type          string        `json:"type,omitempty"`
+}
+
+type Configuration struct {
+	Branches    Branches `json:"branches,omitempty"`
+	GitProvider string   `json:"gitProvider,omitempty"`
+	Path        string   `json:"path,omitempty"`
+}
+
+type Branches struct {
+	Include string `json:"include,omitempty"`
+}
+
+type PipelinesRunID struct {
+	LatestRunID   int    `json:"latestRunId,omitempty"`
+	Name          string `json:"name,omitempty"`
+	SyntaxVersion string `json:"syntaxVersion,omitempty"`
+}
+
+type Console struct {
+	ConsoleID        string      `json:"consoleId,omitempty"`
+	IsSuccess        *bool       `json:"isSuccess,omitempty"`
+	IsShown          interface{} `json:"isShown,omitempty"`
+	ParentConsoleID  string      `json:"parentConsoleId,omitempty"`
+	StepletID        int         `json:"stepletId,omitempty"`
+	PipelineID       int         `json:"pipelineId,omitempty"`
+	Timestamp        int64       `json:"timestamp,omitempty"`
+	TimestampEndedAt interface{} `json:"timestampEndedAt,omitempty"`
+	Type             string      `json:"type,omitempty"`
+	Message          string      `json:"message,omitempty"`
+	CreatedAt        time.Time   `json:"createdAt,omitempty"`
+	UpdatedAt        time.Time   `json:"updatedAt,omitempty"`
+}
+
+// Validation Types
+
+type ValidationResponse struct {
+	IsValid *bool              `json:"isValid,omitempty"`
+	Errors  []ValidationErrors `json:"errors,omitempty"`
+}
+
+type ValidationErrors struct {
+	Text       string `json:"text,omitempty"`
+	LineNumber int    `json:"lineNumber,omitempty"`
+}

pipelines/services/validate.go

@@ -0,0 +1,116 @@
+package services
+
+import (
+	"encoding/json"
+	"errors"
+	"github.com/gookit/color"
+	"github.com/jfrog/jfrog-client-go/auth"
+	"github.com/jfrog/jfrog-client-go/http/jfroghttpclient"
+	"github.com/jfrog/jfrog-client-go/utils/errorutils"
+	"github.com/jfrog/jfrog-client-go/utils/io/httputils"
+	"github.com/jfrog/jfrog-client-go/utils/log"
+	"net/http"
+	"net/url"
+	"strconv"
+)
+
+const (
+	validatePipResourcePath = "api/v1/validateYaml"
+)
+
+type ValidateService struct {
+	client *jfroghttpclient.JfrogHttpClient
+	auth.ServiceDetails
+}
+
+func NewValidateService(client *jfroghttpclient.JfrogHttpClient) *ValidateService {
+	return &ValidateService{client: client}
+}
+
+func (vs *ValidateService) getHttpDetails() httputils.HttpClientDetails {
+	httpDetails := vs.ServiceDetails.CreateHttpClientDetails()
+	return httpDetails
+}
+
+func (vs *ValidateService) ValidatePipeline(data []byte) error {
+	var err error
+	httpDetails := vs.getHttpDetails()
+
+	// Query params
+	m := make(map[string]string, 0)
+
+	// URL Construction
+	uri := vs.constructValidateAPIURL(m, validatePipResourcePath)
+
+	// Headers
+	headers := make(map[string]string)
+	headers["Content-Type"] = "application/json"
+	httpDetails.Headers = headers
+	log.Debug(string(data))
+
+	// Send post request
+	resp, body, err := vs.client.SendPost(uri, data, &httpDetails)
+	if err != nil {
+		return err
+	}
+	if err := errorutils.CheckResponseStatusWithBody(resp, body, http.StatusOK); err != nil {
+		return err
+	}
+
+	// Process response
+	if nil != resp && resp.StatusCode == http.StatusOK {
+		log.Info("Processing resources yaml response ")
+		log.Info(string(body))
+		err = processValidatePipResourceResponse(body)
+		if err != nil {
+			log.Error("Failed to process resources validation response")
+		}
+	}
+	return nil
+}
+
+// processValidatePipResourceResponse process validate pipeline resource response
+func processValidatePipResourceResponse(resp []byte) error {
+	validationResponse := make(map[string]ValidationResponse)
+	err := json.Unmarshal(resp, &validationResponse)
+	if err != nil {
+		return err
+	}
+	if len(validationResponse) == 0 {
+		return errors.New("pipelines not found")
+	}
+	for k, v := range validationResponse {
+		if v.IsValid != nil && *v.IsValid {
+			log.Info("Validation of pipeline resources completed successfully ")
+			msg := color.Green.Sprintf("Validation completed")
+			log.Info(msg)
+		} else {
+			fileName := color.Red.Sprintf("%s", k)
+			log.Error(fileName)
+			validationErrors := v.Errors
+			for _, validationError := range validationErrors {
+				lineNum := validationError.LineNumber
+				errorMessage := color.Red.Sprintf("%s", validationError.Text+":"+strconv.Itoa(lineNum))
+				log.Error(errorMessage)
+			}
+		}
+	}
+	return nil
+}
+
+// constructPipelinesURL creates URL with all required details to make api call
+// like headers, queryParams, apiPath
+func (vs *ValidateService) constructValidateAPIURL(qParams map[string]string, apiPath string) string {
+	uri, err := url.Parse(vs.ServiceDetails.GetUrl() + apiPath)
+	if err != nil {
+		log.Error("Failed to parse pipelines fetch run status url")
+	}
+
+	queryString := uri.Query()
+	for k, v := range qParams {
+		queryString.Set(k, v)
+	}
+	uri.RawQuery = queryString.Encode()
+
+	return uri.String()
+}